Skip to content

observability.md

Latest commit

 

History

History
866 lines (714 loc) · 57.1 KB

observability.md

File metadata and controls

866 lines (714 loc) · 57.1 KB
Raw

moralstack.observability

Unified observability module — single entry point for all telemetry (write + read).

Architecture

moralstack/observability/
├── __init__.py          # exposes: obs (lazy proxy → get_obs()), EventEnvelope, factory helpers
├── events.py            # EventEnvelope dataclass + EVENT_* constants + make_envelope()
├── context.py           # run_id / request_id / cycle contextvars
├── config.py            # getters (get_observability_mode/get_db_path/get_jsonl_dir): explicit > env > default; apply_explicit_config()
├── service.py           # ObservabilityService: emit(), emit_batch(), flush(), shutdown()
├── router.py            # EventRouter: envelope → active sinks (db_only/dual/file_only)
├── read_store.py        # ReadStore Protocol + SqliteReadStore
├── emit_helpers.py      # persist_* / async_persist_* convenience wrappers
├── write_queue.py       # ObservabilityWriteQueue (daemon thread, FIFO)
└── sinks/
    ├── sqlite_sink.py   # Schema SQLite + lifecycle writes + batch inserts
    └── jsonl_sink.py    # Per-event-type JSONL under logs/observability/

The obs export resolves each attribute access on the current get_obs() instance (tests and tooling may replace that singleton). Use get_obs() when you need the concrete ObservabilityService (for example isinstance checks).

Responsibilities

Module Responsibility
events.py EventEnvelope (frozen dataclass), 16 canonical EVENT_* constants, make_envelope() factory
service.py get_obs() thread-safe singleton ObservabilityService; emit() / emit_batch() enqueue, flush(), shutdown(), stats(), read_store property
router.py Reads mode from config; route_window() is counted worker dispatch; route_audit_sync() is counted synchronous finalization dispatch
sqlite_sink.py SQLite schema, SqliteUnitOfWork, synchronous lifecycle writes, FK-ordered write_window() with per-envelope isolation, unique proxy_request_events(run_id, request_id)
jsonl_sink.py Per-event-type JSONL under logs/observability/; thread-safe per-file locks; counted write_window()
read_store.py ReadStore Protocol + SqliteReadStore; unique read contract for UI and reports
emit_helpers.py persist_* / async_persist_* helpers: build EventEnvelope and call get_obs().emit*(); import via moralstack.observability.emit_helpers (not re-exported from package __init__). get_obs binding is deliberately split: async_persist_* use the module-level import; the sync persist_* re-import get_obs lazily inside their try block so a test can patch moralstack.observability.service.get_obs and have it take effect (do not collapse the two — the async contract test patches moralstack.observability.emit_helpers.get_obs)
config.py Env var loader with backwards-compat fallback to legacy vars
context.py ContextVars for run_id, request_id, cycle

Quick start

from moralstack.observability import obs
from moralstack.observability.events import EVENT_LLM_CALL, make_envelope

# Emit a telemetry event (non-blocking)
env = make_envelope(
    EVENT_LLM_CALL,
    run_id="my-run",
    request_id="req-1",
    cycle=1,
    payload={"module": "risk_estimator", "phase": "risk_estimation", ...},
)
obs.emit(env)

# Flush all pending writes (e.g. before shutdown)
obs.flush(timeout=30.0)

# Read back
run = obs.read_store.get_run("my-run")
calls = obs.read_store.get_llm_calls_for_request("my-run", "req-1")

Use obs.emit(...) / obs.emit_batch(...) for queued best-effort telemetry. High-frequency producers should not call router.route(...) / route_batch(...) on the request thread. Decision-audit finalization uses conversation_events.finalize_audit_sync(...), which routes through router.route_audit_sync(...) synchronously and returns a counted result.

EventEnvelope

@dataclass(frozen=True)
class EventEnvelope:
    event_id: str           # UUID4, auto-generated by make_envelope()
    event_type: str         # one of the EVENT_* constants
    timestamp_ms: int       # int(time.time() * 1000), auto-set
    run_id: str | None
    request_id: str | None
    cycle: int | None
    # Multi-turn foundation (mapped to existing DB columns; no schema change)
    session_id: str | None        # -> requests.conversation_id
    turn_number: int | None       # -> requests.turn_index
    parent_event_id: str | None   # -> requests.parent_request_id
    audit_level: str              # "turn" | "session" | "export"
    payload: Mapping[str, Any]    # event-specific data

Canonical event types

Constant String Description
EVENT_RUN_STARTED run.started Run lifecycle start
EVENT_RUN_ENDED run.ended Run lifecycle end
EVENT_REQUEST_UPSERTED request.upserted Request lifecycle
EVENT_REQUEST_DOMAIN_UPDATED request.domain_updated Domain update
EVENT_REQUEST_RESPONSE_UPDATED request.response_updated Response update
EVENT_REQUEST_META_UPDATED request.meta_updated Per-request governance metadata merge (Step 13)
EVENT_LLM_CALL llm.call LLM call telemetry
EVENT_ORCHESTRATION_EVENT orchestration.event Orchestration lifecycle
EVENT_DECISION_TRACE decision.trace Decision trace audit
EVENT_DEBUG_EVENT debug.event Diagnostics / debug
EVENT_CONVERSATION_STATE_UPDATED conversation.state_updated Multi-turn governance state snapshot (Step 13)
EVENT_LEDGER_LOOKUP ledger.lookup Semantic decision ledger lookup (Step 13)
EVENT_LEDGER_STORE ledger.store Semantic decision ledger store (Step 13)
EVENT_SESSION_STORE_GET session_store.get SessionStore get (Step 13)
EVENT_SESSION_STORE_PUT session_store.put SessionStore put (Step 13)
EVENT_PROXY_REQUEST_FINALIZED proxy.request_finalized Per-request proxy finalisation summary (Step 13)

Environment variables

Variable Default Description
MORALSTACK_OBSERVABILITY_MODE db_only if DB_PATH set, else file_only db_only | dual | file_only
MORALSTACK_OBSERVABILITY_DB_PATH — Path to SQLite database
MORALSTACK_OBSERVABILITY_JSONL_DIR logs/observability Directory for JSONL output

Ledger fast-path (Step 14.2)

Variable Default Description
MORALSTACK_LEDGER_ENABLED true Enable the SemanticDecisionLedger semantic cache. When false, every turn runs the full deliberation.
MORALSTACK_LEDGER_SIMILARITY_THRESHOLD 0.92 Cosine similarity threshold for cache hit. Higher = stricter match.
MORALSTACK_LEDGER_MAX_ENTRIES 1000 Maximum cached entries per process (insertion-order (FIFO) eviction beyond this; reads do not refresh).
MORALSTACK_LEDGER_EMBEDDING_MODEL (uses OPENAI_EMBEDDING_MODEL or text-embedding-3-small) Override the embedding model used for similarity.

Skip rules (by design, not configurable):

  • The ledger never caches turns with governance posture ESCALATED (hard-signal refusals).
  • The ledger never caches the very first turn (turn_index < 1).
  • These are safety invariants from the multi-turn design v1.3 section 5.8.

Deprecated aliases (still work, emit a logging.warning):

Old variable New variable
MORALSTACK_PERSIST_MODE MORALSTACK_OBSERVABILITY_MODE
MORALSTACK_DB_PATH MORALSTACK_OBSERVABILITY_DB_PATH

Schema initialization and concurrency

  • init_db() creates the schema and applies the additive migrations idempotently; it never raises and returns a bool.
  • Every SQLite connection goes through _get_connection (sinks/sqlite_sink.py:577-585). On a new file, a concurrent PRAGMA journal_mode=WAL fails at once with database is locked (busy_timeout does not apply to that switch), so the pragma alone is serialized by a per-file lock (_wal_switch_lock, :553-574; key normcase(abspath(path)), lexical, so symlink / 8.3-alias spellings of one file get different locks). It covers every in-process opener (init_db, the write-queue worker, lifecycle writers, readers); the lock is held only for that pragma.
  • Not covered: the same switch racing across processes, and envelopes queued before the first init_db on a fresh / re-pointed file (see docs/CODEBASE_FACTS.md, Future work).
  • DefaultPersistence (orchestration/default_persistence.py) calls init_db lazily, once per DB path: it re-initializes when get_db_path() changes, and retries a failed init on the next call.

Observability modes

db_only (default when DB_PATH is set)

All telemetry is persisted to SQLite. Used by moralstack-ui.

file_only (default when DB_PATH is not set)

All telemetry is written to logs/observability/ as JSONL, one file per event type:

logs/observability/
├── llm.call.jsonl
├── orchestration.event.jsonl
├── decision.trace.jsonl
└── debug.event.jsonl

dual

Both SQLite and JSONL are written simultaneously. Useful for debugging or migration.

For route_window / route_audit_sync accounting, db_only uses SQLite counts, dual uses SQLite as the persisted/failed identity and records JSONL counts separately, and file_only uses JSONL counts. JSONL finalization is synchronously attempted but not crash-durable.

ReadStore

from moralstack.observability import obs

rs = obs.read_store  # SqliteReadStore instance

run        = rs.get_run(run_id)
all_runs   = rs.get_all_runs()
requests   = rs.get_requests_for_run(run_id)
request    = rs.get_request(run_id, request_id)
llm_calls  = rs.get_llm_calls_for_request(run_id, request_id)
token_totals = rs.get_token_usage_totals(run_id, request_id)
token_breakdown = rs.get_token_usage_breakdown(run_id, request_id)
# Per-model rollups (used by the UI token-usage panels at four scopes):
tu_global  = rs.get_token_usage_by_model_global()
tu_run     = rs.get_token_usage_by_model_for_run(run_id)
tu_request = rs.get_token_usage_by_model_for_request(run_id, request_id)
tu_conv    = rs.get_token_usage_by_model_for_conversation(conversation_id)
traces     = rs.get_decision_traces_for_request(run_id, request_id)
orch       = rs.get_orchestration_events_for_request(run_id, request_id)
debug      = rs.get_debug_events_for_request(run_id, request_id)
models     = rs.get_models_used_for_run(run_id)

Model rollup semantics (get_models_used_for_run)

  • The rollup includes only effective llm_calls rows (call_outcome not in skipped, cancelled, discarded).
  • This prevents report headers (for example, "Models used") from showing models that were configured but never actually consumed in the selected run path.
  • risk_estimator rows are mapped by exact action: estimate_intent / estimate_signals / estimate_operational feed risk_intent / risk_signals / risk_operational (first row wins, ordered by started_at). Every risk_estimator row except calibration_guard also feeds the legacy risk key (estimate, retry rows, the minis). calibration_guard rows are synthetic (no provider call) and feed no key. The return type stays dict[str, str].
  • The markdown report (reports/markdown_export.py, _apply_db_risk_models, used by _resolve_models_config_for_run and _overlay_db_models) renders three "MoralStack risk" rows when any mini key exists in the DB or in the benchmark JSON snapshot: a mini key is overwritten only when the DB has it, and shows "—" only when absent from both. A legacy run with only risk keeps the single risk row.
  • The UI run page reads the raw JSON snapshot (ui/templates/run.html:48-96), not the DB overlay, so the UI and the markdown report can disagree for benchmark runs. This is pre-existing and out of scope.

Token usage accounting

Per-request token totals are tracked through two complementary mechanisms:

  1. request_token_usage table — a synchronous, best-effort, potentially partial summary written at request finalization (EVENT_REQUEST_TOKEN_USAGE_FINALIZED). It reflects what the in-process accumulator captured before the HTTP/SDK response was returned. It is not an authoritative or complete guarantee: late async writes, queue drops, or non-billable diagnostic rows can make it diverge from durable llm_calls rows.

    One billable rule applies in-process and in SQL: a falsy billable_provider_call excludes the row, while True, None or an absent key counts it (service._counts_as_billable mirrors COALESCE(billable_provider_call, 1) = 1). The three risk mini-estimator rows carry None and are therefore counted. Comparability: the summary was unaffected by the mini-row gap before 5439aaf (2026-07-18), missing those three calls from 5439aaf until the 2026-10-13 fix, and includes them again after it (it stays best-effort throughout); compare it only within one regime, and across regimes use SUM(llm_calls) over billable rows.

  2. Offline reconstruction from llm_calls — get_token_usage_breakdown() and ad-hoc SUM(...) queries over billable rows (COALESCE(billable_provider_call, 1) = 1) are the most complete source available among rows actually persisted, but still subject to the same best-effort observability queue (rows may be missing if the queue dropped them).

  3. Per-model rollups — get_token_usage_by_model_{global,for_run,for_request, for_conversation}() group the same billable llm_calls rows by model (the conversation variant joins requests.conversation_id). The UI renders these at four scopes (dashboard, run, conversation, single question) through the shared templates/_token_usage.html partial, with estimated/missing badges derived from token_usage_estimated/token_usage_missing.

The OpenAI-compatible proxy usage field mirrors the synchronous accumulator totals on the success path (zeros on fail-closed paths with no generation). Since the 2026-10-13 fix these totals (and the SDK GovernanceMetadata token fields and llm_call_count) include the risk mini-estimators: values rise, shapes are unchanged.

Rows written from worker threads (NEW-1, fixed). async_persist_llm_call drops a row when the run/request id is not in the calling thread's ContextVars. A worker started with a bare ThreadPoolExecutor.submit does not inherit them, so the retry-failed rows of the risk mini-estimators and of the parallel perspectives were silently dropped. Every pool that persists from a worker must submit contextvars.copy_context().run (one copy per task): the estimator pool and _evaluate_parallel now do, like the deliberation modules, the retriever and the speculative overlap. Comparability regime: from the commit that fixed it, llm_call_count, request_token_usage, the SDK/proxy usage and the SQL reconstruction include those failed attempts; earlier runs under-count them. The in-process accumulator and the SQL reconstruction stay in parity by construction (the same emitted row feeds both).

billable_provider_call on llm_calls excludes diagnostic/synthetic rows (speculative reuse, cache hits, skipped modules, leakage-only rows, and the risk estimator's synthetic calibration_guard row) from totals and breakdown queries. Every real provider call must also record its generating model: an unset model collapses into the '—' (empty-key) row of the per-model token panel. All provider-call sites therefore pass the policy/DCCL model explicitly — the fast-path/benign/safe-complete generate (...) calls, the REFUSE (refuse (fast_path)/(deliberative)), compliance-regenerate, draft_revalidate, and the critic/simulator/hindsight/perspectives/risk_estimator retry_failed_attempt_* rows. Synthetic/diagnostic rows that are not provider calls (module skip/gate/disable/error/timeout markers, output-protection leakage rows) intentionally leave model empty and are already billable_provider_call=False.

Fast-path quick_check row (NEW-2, fixed). The critic's HARD quick check is a provider call and is persisted as module="critic", action="quick_check", phase="critic", cycle 0, sequence_in_cycle=2 (after the fast draft, SEQ_POLICY=1), billable_provider_call=True, with the prompt/system prompt sent, the raw reply, the verdict {passed, violated, principle_id} in parsed_summary_json and the billed tokens. A paid reply without a verdict is persisted with error set (passed: null) and the call still raises; a provider error bills nothing and writes nothing. Comparability regime: from that commit one more billable call per fast-path request is counted in llm_call_count and in the request tokens.

Constitution domain-agent calls (module="constitution_retriever", action ∈ domain_prefilter/enhanced_domain_agent/legacy_domain_agent) are billable and counted like any other provider call. The per-domain agents run in a ThreadPoolExecutor; constitution/retriever.py propagates the observability context into each worker via contextvars.copy_context().run, otherwise persist_llm_call would short-circuit on the empty run/request id and drop the row entirely (so their tokens would be missing from every scope).

Who failed (additive keys). The retry_failed rows carry parsed_summary_json keys naming the failed unit: mini_estimator (+estimation_mode) on the estimator rows, perspective_id/perspective_name on the perspectives rows, scenario_id on the hindsight single-path rows; a quick_check row without a verdict has {passed: null, error, error_type}. Existing fields keep their meaning.

Hindsight single-path retry row (NEW-3, fixed). evaluate_scenario writes the same retry_failed row as the batch path (phase="hindsight_retry", prompt Scenario <id> retry reason: ...); same comparability regime.

TokenUsage.from_openai_usage marks a call source="exact" only when the provider returns separate prompt_tokens and completion_tokens. When only total_tokens is available, it estimates the input/output split as 70%/30% of the total and marks source="estimated" — an arbitrary heuristic, not derived from the actual prompt/completion shape. Consumers that need a precise split (not just the total) should check source before trusting input_tokens/output_tokens on estimated rows.

Cached input tokens (prompt-caching observability)

TokenUsage.cached_input_tokens carries the provider-reported subset of input_tokens served from the prompt cache, extracted from usage.prompt_tokens_details.cached_tokens by extract_cached_input_tokens (observability/token_usage.py). It is persisted to llm_calls.cached_input_tokens (nullable, added by the additive migration list) and exposed by get_token_usage_breakdown() as cached_input_tokens plus cached_usage_known.

None and 0 are different answers, and the distinction is load-bearing:

  • None (SQL NULL) — the provider reported no cache details at all. Every row written before this column existed reads as None; so does any non-OpenAI or OpenAI-compatible provider that omits prompt_tokens_details, and every embeddings call (that endpoint has no prompt cache).
  • 0 — the provider measured the call and found no cache hit.

Collapsing the two would make a hit-rate meaningless, so to_json() omits the key when unknown (keeping payloads byte-identical to the pre-caching format) and emits it when it is 0. TokenUsage.combine sums only the known values and stays None when none are known. Extraction is defensive by contract (§5.6): it tolerates a missing or None prompt_tokens_details, a Mapping-shaped one, an Optional[int] cached_tokens of None, and rejects non-int values (a MagicMock coerces to int(1) silently) — it never raises into the caller. Values are clamped to input_tokens.

Because the deliberative modules copy token fields into their own report objects rather than passing the GenerationResult down, cached_prompt_tokens is threaded explicitly through GenerationResult, CriticReport, SimulationResult, HindsightResult, PerspectiveResult and EnsembleResult. Omitting any of those copy sites silently drops the signal for that module while retaining it on its failed-retry rows. Locked by tests/test_cached_token_propagation.py.

Interpreting a hit-rate: OpenAI prompt caching engages only above a minimum prompt length (~1024 tokens) and only on an exact, stable prefix. A 0% hit-rate on short prompts is not evidence that caching is broken — report mean input_tokens and the known/unknown split alongside the ratio. Cached tokens are billed at a reduced rate; they do not reduce the number of tokens consumed.

UI surfaces. The cache hit rate appears everywhere per-module / per-model token metrics already appear:

  • the shared per-model panel (_token_usage_view() → templates/_token_usage.html) at all four scopes (dashboard, run, conversation, single question): a % cached header badge plus Cached in / Cache hit columns per model;
  • the per-module rollup on the request page (_module_summaries());
  • the per-call token_badge macro (a ↻ N badge, shown only on a measured hit);
  • the Domain retrieval table (_domain_retrieval_view()), per call and in total.

All of them render — when the provider reported nothing and 0.0% when it measured a miss; _cache_hit_pct() returns None for the former.

Denominator. The hit rate divides cached tokens by cached_input_base — the input tokens of only the calls the provider reported on — never by total input. A model or module whose rows mix reported and unreported calls (e.g. a database holding runs from before this column existed) would otherwise show a silently diluted percentage. The SQL aggregations expose cached_input_base alongside cached_input_tokens and cached_usage_known.

Locked by tests/test_ui_cache_hit_visibility.py (helpers + end-to-end HTML render, including a legacy-row page that must not print 0%, and explicit no-dilution cases) and tests/test_observability_read_store_token_usage.py.

Deliberately out of scope: the request-level accumulator and request_token_usage do not carry cached tokens (cached ⊆ input, so existing totals stay correct — and that table is a request total, not a per-module/per-model metric), and the proxy's client-facing usage does not expose prompt_tokens_details.

Runtime observability metric contract

Runtime/export builders now include a minimal metric contract (metric_contract) generated from:

  • execution_strategy
  • orchestration_event_count
  • runtime_decisions_rows
  • cycle_cards

Validated invariants include:

  • Speculative timing coherence: 0 <= skip_elapsed_ms <= total_duration_ms (when total duration is available)
  • Activated signals coherence across traces (RISK_ASSESSMENT, PRE_POLICY, FINAL)
  • Row cardinality coherence (runtime_decisions_rows == orchestration_event_count)
  • Path cardinality guardrails (for example, FAST_PATH must not emit cycle cards)

The contract is produced by moralstack.reports.runtime_decisions.build_runtime_observability_contract() and embedded in markdown export structured JSON.

SDK auto-initialisation

When using the Python SDK (govern() / GovernedClient), observability context is initialised automatically — no manual setup is required.

What happens at govern(OpenAI())

  1. GovernedClient.__init__ generates a session-scoped run_id (UUID4) and registers it via set_current_run_id().
  2. If the active mode is db_only or dual, init_db() and create_run() are called to satisfy FK constraints for subsequent event inserts. (Only the WAL switch is serialized in-process; the rest of a concurrent init_db (executescript, migrations, DELETE dedupe) relies on busy_timeout, probe-verified on one platform: see "Schema initialization and concurrency".)
  3. At the end of every chat.completions.create() call via the SDK wrapper, obs.flush() is invoked inside a try/finally block to drain high-frequency queued telemetry for short-lived scripts. The server proxy does not flush per-request (the bounded flush timed out under burst load and only added overhead); it relies on the background worker plus a process-shutdown drain via FastAPI's shutdown hook. Final action metadata and proxy.request_finalized are synchronous on both SDK and proxy via finalize_audit_sync.

Minimum working configuration

# .env
MORALSTACK_OBSERVABILITY_MODE=file_only
MORALSTACK_OBSERVABILITY_JSONL_DIR=./traces/jsonl
from moralstack import govern
from openai import OpenAI

client = govern(OpenAI())
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)
# ./traces/jsonl/llm.call.jsonl, decision.trace.jsonl, etc. are now written.

Observability mode behaviour in SDK

Mode DB init called JSONL written When to use
file_only No Yes Lightweight audit trail, no UI
db_only Yes (if MORALSTACK_OBSERVABILITY_DB_PATH set) No Full UI + query access
dual Yes Yes Debugging / migration

Note: the GovernanceConfig fields observability_mode, db_path, and jsonl_dir are wired through apply_explicit_config (called by govern(), GovernedClient.__init__ and create_app). Precedence is explicit > env (MORALSTACK_OBSERVABILITY_*) > default; "off"/blank/None inherit env/defaults and do not disable writing. Conflict policy: explicit values are process-wide and only establish the target, i.e. they apply only if the first call in the process happens before the write-queue target is bound (ObservabilityWriteQueue.target_bound, exposed by ObservabilityService.target_bound; set on the first enqueue or worker connection open; merely creating the singleton, e.g. via import moralstack.reports, does not bind it). A later different value is ignored with a cause-specific WARNING (once per key and value) and never redirects a live target: the worker's SQLite connection is opened once while parent rows are written synchronously to get_db_path(), so a redirect would split parents and children across databases. A govern() that fails after the apply step (e.g. a pipeline build error) still fixes the target for the process. Residual race: _enqueue does not take _obs_lock, so a non-SDK emitter can enqueue on an already-existing singleton between the bound check and the publish in _try_establish; no SDK/proxy client exists yet while _applied is false, so the window is accepted. Getters read env live on every call and never raise.

Context vars

from moralstack.observability.context import (
    get_current_run_id, set_current_run_id,
    get_current_request_id, set_current_request_id,
    get_current_cycle, set_current_cycle,
)

These are propagated across thread boundaries via contextvars.copy_context() inside ObservabilityWriteQueue.submit().

In the SDK path, run_id is set automatically by GovernedClient._init_run_context(). Direct calls to set_current_run_id() are only needed when using the orchestrator API directly (e.g. CLI or custom integrations).

Multi-turn conversation observability (Step 13)

MoralStack treats every chat as a sequence of governed turns. Step 13 adds first-class persistence and read APIs for the conversation-level signals that previously lived only in transient runtime state.

New SQLite tables

Table Written by Purpose
conversation_states EVENT_CONVERSATION_STATE_UPDATED Full state_in / state_out snapshots + summary per turn
ledger_events EVENT_LEDGER_LOOKUP, EVENT_LEDGER_STORE Every semantic decision ledger lookup/store with similarity, outcome, reason
session_store_events EVENT_SESSION_STORE_GET, EVENT_SESSION_STORE_PUT Every SessionStore access with TTL / eviction metadata
proxy_request_events EVENT_PROXY_REQUEST_FINALIZED Per-request proxy finalisation summary (state_provided, state_updated, postures, was_cached, response length, headers)

The existing requests table is extended with a meta_json column merged turn-by-turn via EVENT_REQUEST_META_UPDATED (final_action, risk_score, path, posture, reason codes, triggered principles, …).

Emitters

from moralstack.observability.conversation_events import (
    emit_request_meta_updated,
    emit_conversation_state_updated,
    emit_ledger_lookup,
    emit_ledger_store,
    emit_session_store_get,
    emit_session_store_put,
    emit_proxy_request_finalized,
)

Most emitters are best-effort: they swallow failures and enqueue without blocking the hot governance path. The finalization helper finalize_audit_sync() is the synchronous audit exception: it writes request.meta_updated and proxy.request_finalized through route_audit_sync(), counts failures in the returned result and obs.stats(), and still never raises into the caller.

Read API

rs = obs.read_store
rs.get_requests_for_conversation(conversation_id)
rs.get_conversation_states(conversation_id)
rs.get_ledger_events_for_conversation(conversation_id)
rs.get_session_store_events_for_conversation(conversation_id)
rs.get_proxy_request_events_for_conversation(conversation_id)
rs.get_conversation_ids_for_run(run_id)
rs.get_conversation_overview(conversation_id)

get_conversation_overview() returns aggregated metrics keyed by turn_count, first_created_at, last_created_at, final_actions, max_risk_score, last_posture, ledger_hits, ledger_misses, session_store_hits, session_store_misses, any_turn_cached, cached_turn_count.

Orchestration event sub-types

Each orchestration.event envelope carries an internal event_type field (distinct from the envelope-level event_type) that identifies the kind of orchestration step. The full taxonomy is in moralstack/orchestration/orchestration_event_taxonomy.py. Notable types relevant to multi-turn observability:

Sub-type Stage Component Meaning
CONVERSATION_CONTEXT_ATTACHED orchestration conversation Conversation context was attached to the request (session id, turn index, parent request).
CONVERSATION_STATE_UPDATED orchestration conversation The ConversationGovernanceState was advanced after this turn.
LEDGER_FAST_PATH_APPLIED fast_path ledger_fast_path_runner The SemanticDecisionLedger returned a hit AND the safety gate accepted it; deliberation was skipped. Payload: from_turn, similarity, cached_action, forced_route, modules_skipped.
LEDGER_FAST_PATH_NOT_APPLIED fast_path ledger_fast_path_runner The SemanticDecisionLedger returned a hit but it was not applied: either the safety gate refused it (the current route requires deliberation and the cached decision is not REFUSE: gate_reason current_route_requires_deliberation) or the hard-signal guard did (the request carries hard-signal evidence, gate_reason hard_signal_evidence, or is a hard-signal REFUSE, hard_signal_refuse). The turn continues on its current route as on a cache miss. Payload: from_turn, similarity, cached_action, current_route, gate_reason.

Fast-path safety gate

The SemanticDecisionLedger cache is NOT applied unconditionally on a hit. The ConversationalFastPathRunner.is_safe_to_apply gate enforces a safety invariant: a cached decision is applied only when applying it would not weaken safety relative to the current run.

Rules:

  1. Cached REFUSE → always applied. A cached refusal is always safe to reuse: the system already decided to deny on a semantically-equivalent query.
  2. Non-REFUSE on non-deliberative current route → applied. When the current run's path router returns one of benign, safe_complete, refuse, or fast_path, the current run has not requested deliberation. Reusing the cached non-REFUSE decision is safe.
  3. Non-REFUSE on deliberative current route → REJECTED. When the current run's path router returns deliberative or deliberative_loop, the current turn has a risk profile that triggers full deliberation. Reusing the cached (less strict) decision would downgrade safety — the gate refuses and the deliberation runs in full.
  4. Non-REFUSE with hard-signal evidence → REJECTED (any route). The gate above is risk-blind, so the controller applies a second rule (ledger_hard_signal_block_reason, PROJECT_SPEC §5.3): a cached NORMAL/SAFE decision is never reused when has_hard_signal_evidence holds for the current risk estimation (hard_signal_evidence) or the current decision is a hard-signal REFUSE (hard_signal_refuse, defense in depth for a ledger that ignores the ESCALATED posture). A cached REFUSE is never blocked. The request continues exactly as on a cache miss; on fast_path that means the HARD quick_check of the fresh draft still runs.

Observability:

  • When the gate applies the hit, an orchestration.event with event_type='LEDGER_FAST_PATH_APPLIED' is emitted.
  • When the gate rejects the hit, an orchestration.event with event_type='LEDGER_FAST_PATH_NOT_APPLIED' is emitted, with reason_codes=['current_route_requires_deliberation'] (or ['hard_signal_evidence'] / ['hard_signal_refuse'] for rule 4) and payload.gate_reason indicating the rejection cause.

Demo: examples/multiturn_quickstart_gate_rejected.py reproduces a gate-rejected scenario (best-effort, depends on the risk estimator classifying the turn as deliberation-worthy).

UI surfaces (moralstack-ui)

  • GET /conversations/{conversation_id} — full timeline with per-turn governance decision, state transitions, ledger and session-store activity, and proxy finalisation.
  • GET /conversations/{conversation_id}/export.md — Markdown audit trail (moralstack.reports.conversation_export.export_conversation_to_markdown).
  • GET /conversations?q=<id> — direct lookup redirect.
  • The run detail page now lists every conversation seen in the run; the request detail page surfaces the parent conversation context.

CLI inspector

python scripts/inspect_multiturn_trace.py <conversation_id>          # human summary
python scripts/inspect_multiturn_trace.py --list-run <run_id>        # list conversations
python scripts/inspect_multiturn_trace.py <conversation_id> --json   # raw rows
python scripts/inspect_multiturn_trace.py <conversation_id> --export trace.md

Requires MORALSTACK_OBSERVABILITY_DB_PATH (or the legacy MORALSTACK_DB_PATH) to point at a populated SQLite database.

JSONL semantics: atomic vs merge vs upsert

The JSONL stream is append-only: every envelope produces one new line in the matching <event_type>.jsonl file. The SQLite sink, in contrast, applies different write strategies depending on the event_type. Consumers that read the JSONL offline must understand which event types are deltas and which are self-contained snapshots, otherwise they will misinterpret the data.

Three categories of event semantics

Category Event types JSONL contains SQLite write strategy
Atomic insert llm.call, orchestration.event, decision.trace, debug.event, ledger.lookup, ledger.store, session_store.get, session_store.put, proxy.request_finalized one independent record per envelope; never merged with previous entries one new row per envelope (INSERT INTO ...)
Merge-update request.meta_updated, request.domain_updated, request.response_updated a delta: the envelope's payload.meta (or domain, or final_response) is a partial update, not the final state the existing row is updated with a JSON merge (update_request_meta(merge=True)); the SQLite column always holds the consolidated state
Snapshot (first-writer or append) request.upserted, conversation.state_updated, run.started, run.ended a complete snapshot of the entity at envelope time per type (code wins, sqlite_sink.py): run.started is INSERT OR IGNORE (first writer wins, :878); request.upserted is INSERT OR IGNORE (:957); run.ended is an UPDATE of the run row (:911-913); conversation.state_updated is a plain INSERT (:1175-1176, one row per envelope)

Concrete consequence for offline consumers

If you read logs/observability/request.meta_updated.jsonl with a script that takes the last line for a given (run_id, request_id) pair, you will see only the last delta — not the consolidated meta_json that SQLite holds. To reconstruct the consolidated state from the JSONL alone, you must:

  1. read all envelopes for the target (run_id, request_id),
  2. apply them in created_at order,
  3. merge each payload.meta into an accumulator (last-write-wins per key).

Atomic-insert events do not have this problem: every envelope is already self-contained.

Snapshot events are whole-payload records, not per-key deltas. In SQLite the strategy depends on the type (see the table): for run.started and request.upserted the first writer wins; run.ended updates the existing run row; conversation.state_updated appends one row per envelope. There is no INSERT OR REPLACE.

Prompt-set audit (Phase A, step A4)

  • Run level. runs.meta_json.prompt_config carries PromptSet.audit_summary() (schema 1: set and default fingerprints, ps, is_default, prompts_dir as configured, override_files with their own sha256, safety_fragments_overridden, per-fragment sha256/when/origin, overridden text, missing_placeholders). The key is added to the existing meta of every entry point (SDK sdk_session, proxy proxy, CLI single, benchmark benchmark) and to the session fallback (DefaultPersistence(run_meta=...), session). Roughly 21 KB for the default set, plus override texts. Nothing is added when the orchestrator is not a real PromptSet holder (mocks) or when the summary cannot be built or JSON-serialized (prompt_run_meta returns {}, DEBUG log).
  • File record. conversation_events.emit_run_started_file_record(run_id, run_type, meta) writes ONE canonical run.started envelope (payload={"run_type", "meta"}) to run.started.jsonl through router.route_jsonl (JSONL sink only, file_only/dual only, synchronous, never raises). It is JSONL-only because the DB row is created by the entry point's direct create_run; routing run.started through SQLite too would create the run twice. A process-wide, lock-guarded, bounded (4096, insertion-ordered) once-per-run-id guard mirrors the DB's first-writer semantics: an entry point's sdk_session line is never followed by a session line for the same run id. DefaultPersistence.ensure_run_and_upsert_request emits it before its no-DB early return, so library callers in file_only get the record too. The CLI and the benchmark now also set a run id when the mode is file_only (with a DB path they already did; dual without a DB path still sets none). In file_only the benchmark also writes its benchmark_<run_id>.json report, which the UI cannot reach (no runs row).
  • Request link. OrchestrationController.process emits request.meta_updated with meta={"prompt_set": {"ps", "is_default"}} right after the request upsert (merge into requests.meta_json; a delta line in request.meta_updated.jsonl; no-op without a run id; a missing row is a counted window failure). It identifies the set, not the per-call when variant (per-call identity comes with A4b's wire hash).
  • Override texts are persisted verbatim in both channels: never put secrets in prompts.

Wire record (Phase A, step A4b)

The llm_calls columns (prompt, system_prompt) do not always hold what the model actually received: the module context preface, the developer-contract and history messages, the retry suffix and the legacy generate() branch are not in them, and several calls share one row. The wire record closes that without changing the meaning of any existing field: it describes the exact messages list sent, storing only what the columns lack, plus a sha256 of the whole list.

  • Where. parsed_summary_json["wire"] for a row that records one call. A row that aggregates several calls (perspectives, a hindsight row that evaluated several consequences one by one, the refusal's two attempts) uses parsed_summary_json["wire_calls"] (records in call order) and parsed_summary_json["wire_strings"] (one string table shared by all of them, so a system prompt common to every call is stored once). A cache hit makes no call and carries no wire. The record travels in the same llm.call envelope to SQLite and to llm.call.jsonl.

  • Record v1.

    {"v": 1, "branch": "messages|generate", "sha256": "...", "n": 4,
 "messages": [{"role": "system", "ref": "system_prompt"},
              {"role": "developer", "s": 0},
              {"role": "user", "s": 1},
              {"role": "user", "ref": "prompt", "prefix": "<preface>", "suffix": ""}],
 "strings": ["<contract as sent>", "<history turn as sent, truncated>"]}

    Each message content is ref (byte-equal to the row's system_prompt or prompt column), ref with prefix/suffix (the column occurs exactly once inside the content: the preface and the retry suffix wrap the persisted prompt), or s (an index into the string table, which deduplicates identical contents). branch is messages when the call went through generate_messages and generate for the legacy generate() call. sha256 is the digest of json.dumps([{"role", "content"}, ...], ensure_ascii=False, separators=(",", ":")), encoded as UTF-8 with errors="surrogatepass" (a lone surrogate hashes instead of raising). A verifier must call wire_sha256 rather than re-implement the digest. decode_wire raises ValueError (and only ValueError) on any malformed record, including a missing key or an out-of-range string index.

  • Codec. moralstack/observability/wire.py is pure (no I/O): generate_messages_for(prompt, system) mirrors OpenAIPolicy.generate (the system message is omitted when empty), encode_wire, encode_wire_calls, decode_wire and wire_sha256. decode_wire(encode_wire(m, ...)) == m and the digest of the decoded messages equals the record's sha256 (tests/test_wire_codec.py, including {, {{, CRLF, NBSP and empty strings). Building a record is telemetry: it is built at the call site from the exact arguments passed to the policy, wrapped like every other emit, and a failure never reaches the module call.

  • message_sections keeps its meaning. It is the request's context (developer messages, history, final user message) stored in full, in the row's summary or in the module's CONTEXT_SHAPE_RECORDED event. wire holds the bytes actually sent (windowed, truncated, with the preface). Both are kept.

  • Not changed, on purpose. prompt does not start to include the preface, message_sections is not truncated to the window sent, the "Retry reason: ..." prompt of retry-failed rows stays (the failed attempt's wire is added next to it, with its system prompt inlined because those rows have no system_prompt), and aggregated rows are not split.

  • Status. A4b is complete: every chat-completion LLM-call row carries the record, and KNOWN_GAPS in tests/test_persistence_fidelity.py is empty. Two rows carry none by design: embedder/embed (an embeddings call, not a chat completion) and upstream_provider/generate final response (no caller records it). Batch 2: estimator mini rows, critic, simulator, hindsight and perspectives rows (and their retry-failed rows). Batch 3: the DCCL evaluate, safety_override and draft_revalidate rows; the contract-regeneration (generate (compliance-regenerate)) row, which has no system_prompt column so its wire inlines the system message; the speculative draft row (internal and upstream); the policy rows (generate on the benign / fast / safe-complete / deliberative paths, rewrite and soft revision); the critic quick_check row; the refusal rows (wire_calls, one record per attempt, on the existing row); the constitution retriever rows (prefilter, enhanced and legacy domain agents). A new row is persisted for the domain-exclusion message call (module="orchestration", action="domain_excluded", phase="domain_exclusion", call_kind="domain_exclusion"), which was not recorded before. A domain-exclusion call that raises leaves no row (unlike refusal, which records the attempt): the response falls back to the marker and nothing is persisted for the failed call; DeveloperContractComplianceLayer.validate_contract now classifies through _classify_safety_override, so its classifier call is persisted too (only when it runs inside a request scope: at contract-loading time there is none, so nothing is written).

  • Columns that differ from the wire on purpose. Two rows keep a prompt column that is not among the bytes sent, because changing it would change an existing field: the DCCL row on the legacy generate() branch (it keeps the native trailer, the fallback user prompt was sent) and the speculative row on the native branch (it keeps the language-prefixed prompt, the raw final user message was sent). The wire is the authority for both; test_persistence_fidelity.py lists them in _PROMPT_COLUMN_DIVERGES_BY_DESIGN. A third case is not exempted because the shipped adapters cannot reach it: a speculative row on the legacy branch with a developer contract encodes the system message as a ref plus a prefix (the contract is prepended to the system prompt), so the decoded system differs from the system_prompt column. Likewise, prefix/suffix refs and inlined system messages (the G13 contract-regeneration row and the retry-failed rows, which have no system_prompt column) are by design: decoded system == system_prompt column is not an integrity rule, the integrity rule is that the decoded record equals the messages sent.

  • UI effect. Policy, refusal, draft-match and G13 rows, whose parsed_summary_json was NULL before, now carry a JSON summary, so the request view shows a raw wire JSON block and a wire: {...} preview on their cards. No UI code changed. rewrite rows record the adapter's own prompt_used / system_used (the rewrite prompt is composed inside the policy) and carry no wire when the adapter recorded none.

  • Where each module puts it. Estimator mini rows, simulator, hindsight and perspectives rows: parsed_summary_json (wire, or wire_calls + wire_strings for the perspectives row and a multi-consequence hindsight row). Critic rows: parsed_json ({"wire": {...}}), because the critic's parsed_summary_json is the plain text Violations: N, Guidance: ... that tests pin; it stays untouched. Retry-failed rows of every module carry an inline wire in their summary (the failed attempt's messages, system prompt inlined; the row has no system_prompt). The perspectives row lists the successful calls, in the order of its joined prompts. The module result objects carry the record in an additive optional field (CriticReport.wire, SimulationResult.wire, HindsightResult.wire/wire_calls, EnsembleResult.wire_calls, RefusalGenerationResult.wire_calls); a cache hit clears it. Batch-3 rows whose summary is a JSON object (DCCL, safety classifier, speculative, critic quick_check) add wire next to their existing keys; rows with no summary before (policy, draft-match, contract regeneration, refusal, domain exclusion) get a summary holding only the wire keys (wire_summary_json); retriever rows add wire to the summary that already holds parse_contract. That test drives every module and path with recording fake policies against a real SQLite DB in dual mode, reconstructs each call from its row and from llm.call.jsonl, and asserts the gap report equals KNOWN_GAPS: a new gap fails it, and so does a fixed gap that is still listed.

Prompt-audit verifier (Phase A, step A4c)

scripts/verify_prompt_audit.py is a read-only, offline, free check that the prompt audit is consistent with the packaged catalog. It can be run on any existing database (a pre-feature DB is classified, not rejected).

python scripts/verify_prompt_audit.py --db moralstack.db [--run-id ID ...] [--jsonl-dir DIR] [--json] [--strict]
  • Read-only. SQLite is opened with file:<path>?mode=ro plus PRAGMA query_only; JSONL files are only read. No network, no LLM call. On a WAL-mode database mode=ro may still create the -wal / -shm sidecar files when it reads (the main database file's bytes, size and mtime stay unchanged; run it on a copy if even that matters). immutable=1 is deliberately not used: it skips locking and ignores the WAL, so it would read a stale or torn view of a live database. The script imports only moralstack.prompts.catalog, moralstack.observability.wire and the risk signal registry/renderer, and it registers moralstack.observability and moralstack.models.risk(.signals) as empty placeholder packages first (inside the script only; no production change) so their __init__ never runs: the risk estimator, moralstack.orchestration and the observability service (queue, sinks) are not loaded, nor any HTTP library, thread or logging handler. --jsonl-dir alone makes the JSONL channel the data source (llm.call.jsonl, run.started.jsonl, request.upserted.jsonl, request.meta_updated.jsonl, plus rotated <event>.*.jsonl; other files such as the file_only benchmark orphan benchmark_<run_id>.json are ignored); with --db it only supplies the run prompt_config and the request link when the DB lacks them. A second run.started line for the same run id (written after the guard evicted the id) is ignored: the first line wins.
  • Per run. prompt_config is read from runs.meta_json (or run.started), checked for its schema keys and hash formats, and default_fingerprint is compared with the packaged defaults. If they match, the effective set is rebuilt as the default set plus the recorded per-fragment overridden text / when (through the public loader on a temporary prompts_dir), and both set_fingerprint and every per-fragment sha256 / when must reproduce (config_status: mismatch otherwise). If the defaults differ, the run is defaults_differ and its rows are unknown (not re-rendered). safety_fragments_overridden is listed per run. A run without prompt_config is no prompt_config (pre-feature) and its rows are checked against the current packaged defaults.
  • Per row. Each llm_calls row (or llm.call envelope) gets one status. The code checks in this order: synthetic first; then a wire record that fails integrity (wire_integrity); then, if the run's set is not checkable or the (module, action) has no matcher, unknown; then a row with no system prompt at all (no_wire without a wire record, missing with one); then mismatch when a recorded system prompt is not an expected one; otherwise ok.
    • synthetic: not a chat completion, no record by design (orchestration notes, reused drafts, leakage_detected*, timeout_warning, pre_cycle_check, calibration guard, * (SKIPPED|GATED|DISABLED|ERROR), embedder/embed, upstream_provider/generate final response, billable_provider_call = 0, a refusal with attempts = 0).
    • wire_integrity: a wire / wire_calls record (holder: the summary when it holds one, else parsed_json, which is where the critic keeps it) does not decode, an n or a ref does not resolve (every ref needs a non-empty column), or wire_sha256(decode_wire(...)) differs from the record's sha256. Each record of wire_calls is checked separately against the shared wire_strings.
    • mismatch / ok: the row's system_prompt column, and the leading system message of every wire record, must each be one the effective set renders for (module, action). Context-aware compositions are tried for all four contexts; the policy system prompt is checked structurally per action (optional contract prefix, policy.system + language rules + anti-leak rules + the mandatory §REF_<8 hex>§ canary line; rewrite / soft_revision must then end with policy.rewrite.constraints (or be the rewrite default system plus the constraints), generate actions must not carry them; the runtime never puts a safe-complete / constrained instruction in the system prompt, so it is not accepted); the prefilter system prompt is matched as its template; only the perspectives row aggregates its system column (per-call systems joined by \n---\n, matched piece by piece). Retry-failed rows, whose system prompt is inlined in the wire and whose column is empty, are checked on the wire system message.
    • unknown: no matcher for the (module, action), or the run's set cannot be re-rendered.
    • no_wire: no system prompt recorded and no wire record (pre-feature rows, retry-failed rows written before the wire).
    • missing: a wire record exists but carries no system message and the column is empty.
  • Informational, never an error. Column equality (system_prompt equals the decoded system message, prompt is among the decoded user messages) is false by design for the G13 and retry-failed rows, dccl/generate, speculative/messages, the preface and retry-suffix rows and any prefix/suffix ref; it is reported as a count. branch counts are reported and never failed on: branch=generate is only proven for OpenAIPolicy and UpstreamDraftGenerator, and rewrite / soft_revision wires are adapter-attested (the adapter's own prompt_used / system_used).
  • Per request. requests.meta_json.prompt_set.ps must equal the run's prompt_config.ps (ok, missing when the link is absent, mismatch); in a pre-feature run requests are counted as pre-feature. In JSONL mode the request side is scripts/consolidate_jsonl_meta.consolidate over request.upserted and request.meta_updated.
  • Output and exit code. A per-run summary, then a per-module table of ok / missing / mismatch / unknown / synthetic / no_wire / wire_integrity counts with the offending row ids (and with_wire); --json prints the same plus the legend. Each offending row id comes with its reason (--json: offending_row_details, keyed by row id; text: a short reason next to each id, capped). Row ids are llm_calls.id with --db and the envelope event_id in JSONL mode. Exit code 0; with --strict, 1 when any row is a mismatch or wire_integrity, or any run (mismatch, malformed, unrebuildable) or request is a mismatch (a missing link alone does not fail); 2 on an unreadable input or a usage error. An unrebuildable run (the recorded overrides do not form a valid set) fails strict, while defaults_differ (an older packaged-defaults version) does not.
  • DB versus JSONL status (N8). A run created with empty meta has no prompt_config in runs.meta_json; the DB-only run is then pre-feature, while the same run read from run.started.jsonl (or from --db plus --jsonl-dir, which falls back to it) may carry one. The two modes can therefore classify the same run differently, by design.
  • What ok means. "What the effective catalog would have produced", for the shapes the tool knows. It does not prove the model received it for rows without a wire record, and it says nothing about a (module, action) it has no matcher for (unknown). Tests: tests/test_verify_prompt_audit.py.

Worked example: request.meta_updated

A single request goes through five state transitions, producing five envelopes in the JSONL:

{"event_type":"request.meta_updated","run_id":"r1","request_id":"req1","payload":{"meta":{"final_action":"NORMAL_COMPLETE","path":"FAST_PATH"}}}
{"event_type":"request.meta_updated","run_id":"r1","request_id":"req1","payload":{"meta":{"risk_score":0.10}}}
{"event_type":"request.meta_updated","run_id":"r1","request_id":"req1","payload":{"meta":{"intent_clarity":"HIGH"}}}
{"event_type":"request.meta_updated","run_id":"r1","request_id":"req1","payload":{"meta":{"was_cached":true,"cached_from_turn":1}}}
{"event_type":"request.meta_updated","run_id":"r1","request_id":"req1","payload":{"meta":{"governance_posture":"ELEVATED"}}}

The SQLite row for that request, after the writer has processed all five:

{
  "final_action": "NORMAL_COMPLETE",
  "path": "FAST_PATH",
  "risk_score": 0.10,
  "intent_clarity": "HIGH",
  "was_cached": true,
  "cached_from_turn": 1,
  "governance_posture": "ELEVATED"
}

That consolidated form is what obs.read_store.get_request(run_id, request_id) returns and what moralstack-ui displays. The five JSONL lines, taken individually, never contain the full picture.

Helper script

scripts/consolidate_jsonl_meta.py performs the JSONL→consolidated merge offline. Use it when you need the SQLite-equivalent state but only have the JSONL files (for example: when copying logs between machines, when auditing after the DB has been rotated, or when feeding offline analytics that work with self-contained JSON).

python scripts/consolidate_jsonl_meta.py \
    --input logs/observability/request.meta_updated.jsonl \
    --output requests_meta_consolidated.json

The output is a JSON file mapping (run_id, request_id) → consolidated meta dict, with the same semantics as requests.meta_json in SQLite.