feat: Runtime Health Event Pipeline - Waves 0-2 [OMN-5529]

.github/workflows/test.yml

@@ -601,7 +601,7 @@ jobs:
         run: |
           uv run pytest tests/ \
             --ignore=tests/integration/docker \
-            -m "not slow and not chaos" \
+            -m "not slow and not chaos and not kafka" \
             --splits 10 \
             --group ${{ matrix.split }} \
             -n auto \

docker/migrations/forward/054_create_consumer_health_triage.sql

@@ -0,0 +1,111 @@
+-- =============================================================================
+-- MIGRATION: Create consumer_health_triage and consumer_restart_state tables
+-- =============================================================================
+-- Ticket: OMN-5512 (Task 7: Create migration for consumer_health_triage table)
+-- Epic: OMN-5529 (Runtime Health Event Pipeline)
+-- Version: 1.0.0
+--
+-- PURPOSE:
+--   Creates the consumer health triage tables for Layer 1 of the runtime
+--   health event pipeline. consumer_health_triage tracks incidents by
+--   fingerprint for graduated response. consumer_restart_state tracks
+--   restart rate limiting by consumer identity.
+--
+-- IDEMPOTENCY:
+--   - CREATE TABLE IF NOT EXISTS is safe to re-run.
+--   - CREATE INDEX IF NOT EXISTS is safe to re-run.
+--
+-- ROLLBACK:
+--   See rollback/rollback_054_create_consumer_health_triage.sql
+-- =============================================================================
+
+-- Incident tracking by fingerprint for graduated response
+CREATE TABLE IF NOT EXISTS public.consumer_health_triage (
+    id                      BIGSERIAL       PRIMARY KEY,
+    fingerprint             TEXT            NOT NULL,
+    consumer_id             TEXT            NOT NULL,
+    consumer_group          TEXT            NOT NULL,
+    topic                   TEXT            NOT NULL,
+    event_type              TEXT            NOT NULL,
+    severity                TEXT            NOT NULL,
+    incident_state          TEXT            NOT NULL DEFAULT 'open',
+    occurrence_count        INT             NOT NULL DEFAULT 1,
+    first_seen_at           TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
+    last_seen_at            TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
+    escalated_at            TIMESTAMPTZ,
+    resolved_at             TIMESTAMPTZ,
+    linear_ticket_id        TEXT,
+    error_message           TEXT            DEFAULT '',
+    service_name            TEXT            DEFAULT '',
+    hostname                TEXT            DEFAULT '',
+    correlation_id          TEXT
+);
+
+-- Restart rate limiting by consumer identity
+CREATE TABLE IF NOT EXISTS public.consumer_restart_state (
+    id                      BIGSERIAL       PRIMARY KEY,
+    consumer_id             TEXT            NOT NULL,
+    consumer_group          TEXT            NOT NULL,
+    topic                   TEXT            NOT NULL,
+    last_restart_at         TIMESTAMPTZ,
+    restart_count_30min     INT             NOT NULL DEFAULT 0,
+    restart_window_start    TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
+    last_restart_success    BOOLEAN,
+    created_at              TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
+    updated_at              TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
+    UNIQUE (consumer_id, consumer_group, topic)
+);
+
+-- =============================================================================
+-- INDEXES
+-- =============================================================================
+
+-- Triage table indexes
+CREATE INDEX IF NOT EXISTS idx_consumer_health_triage_fingerprint
+    ON consumer_health_triage (fingerprint);
+
+CREATE INDEX IF NOT EXISTS idx_consumer_health_triage_consumer
+    ON consumer_health_triage (consumer_id, consumer_group);
+
+CREATE INDEX IF NOT EXISTS idx_consumer_health_triage_state
+    ON consumer_health_triage (incident_state);
+
+CREATE INDEX IF NOT EXISTS idx_consumer_health_triage_last_seen
+    ON consumer_health_triage (last_seen_at);
+
+CREATE INDEX IF NOT EXISTS idx_consumer_health_triage_fingerprint_state
+    ON consumer_health_triage (fingerprint, incident_state);
+
+-- Restart state indexes
+CREATE INDEX IF NOT EXISTS idx_consumer_restart_state_consumer
+    ON consumer_restart_state (consumer_id, consumer_group, topic);
+
+-- =============================================================================
+-- COMMENTS
+-- =============================================================================
+
+COMMENT ON TABLE consumer_health_triage IS
+    'Incident tracking for consumer health events (OMN-5512). '
+    'Each row tracks an incident by fingerprint for graduated response: '
+    'Slack warning -> Slack repeated -> restart command -> Linear ticket.';
+
+COMMENT ON COLUMN consumer_health_triage.fingerprint IS
+    'Stable hash of (consumer_id, event_type, topic) for deduplication.';
+
+COMMENT ON COLUMN consumer_health_triage.incident_state IS
+    'Lifecycle state: open, acknowledged, restart_pending, restart_succeeded, '
+    'restart_failed, ticketed, resolved.';
+
+COMMENT ON COLUMN consumer_health_triage.occurrence_count IS
+    'Number of times this fingerprint has been seen in the current incident.';
+
+COMMENT ON TABLE consumer_restart_state IS
+    'Restart rate limiting state per consumer identity (OMN-5512). '
+    'Prevents restart storms by tracking restart frequency within a 30-min window.';
+
+-- Update migration sentinel
+UPDATE public.db_metadata
+SET migrations_complete = TRUE,
+    schema_version = '054',
+    updated_at = NOW()
+WHERE id = TRUE AND owner_service = 'omnibase_infra';

docker/migrations/forward/055_create_runtime_error_triage.sql

@@ -0,0 +1,93 @@
+-- =============================================================================
+-- MIGRATION: Create runtime_error_triage table
+-- =============================================================================
+-- Ticket: OMN-5514 (Task 13: Create migration for runtime_error_triage table)
+-- Epic: OMN-5529 (Runtime Health Event Pipeline)
+-- Version: 1.0.0
+--
+-- PURPOSE:
+--   Creates the runtime error triage table for Layer 2 of the runtime
+--   health event pipeline. Tracks runtime error incidents by fingerprint
+--   for graduated triage response with cross-layer correlation to Layer 1
+--   consumer health incidents.
+--
+-- IDEMPOTENCY:
+--   - CREATE TABLE IF NOT EXISTS is safe to re-run.
+--   - CREATE INDEX IF NOT EXISTS is safe to re-run.
+--
+-- ROLLBACK:
+--   See rollback/rollback_055_create_runtime_error_triage.sql
+-- =============================================================================
+
+CREATE TABLE IF NOT EXISTS public.runtime_error_triage (
+    id                      BIGSERIAL       PRIMARY KEY,
+    fingerprint             TEXT            NOT NULL,
+    logger_name             TEXT            NOT NULL,
+    error_category          TEXT            NOT NULL,
+    severity                TEXT            NOT NULL,
+    incident_state          TEXT            NOT NULL DEFAULT 'open',
+    occurrence_count        INT             NOT NULL DEFAULT 1,
+    message_template        TEXT            NOT NULL DEFAULT '',
+    first_seen_at           TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
+    last_seen_at            TIMESTAMPTZ     NOT NULL DEFAULT NOW(),
+    escalated_at            TIMESTAMPTZ,
+    resolved_at             TIMESTAMPTZ,
+    linear_ticket_id        TEXT,
+    -- Cross-layer correlation: links to consumer_health_triage fingerprint
+    correlated_consumer_fingerprint TEXT,
+    service_name            TEXT            DEFAULT '',
+    hostname                TEXT            DEFAULT '',
+    correlation_id          TEXT
+);
+
+-- =============================================================================
+-- INDEXES
+-- =============================================================================
+
+CREATE INDEX IF NOT EXISTS idx_runtime_error_triage_fingerprint
+    ON runtime_error_triage (fingerprint);
+
+CREATE INDEX IF NOT EXISTS idx_runtime_error_triage_category
+    ON runtime_error_triage (error_category);
+
+CREATE INDEX IF NOT EXISTS idx_runtime_error_triage_state
+    ON runtime_error_triage (incident_state);
+
+CREATE INDEX IF NOT EXISTS idx_runtime_error_triage_last_seen
+    ON runtime_error_triage (last_seen_at);
+
+CREATE INDEX IF NOT EXISTS idx_runtime_error_triage_fingerprint_state
+    ON runtime_error_triage (fingerprint, incident_state);
+
+CREATE INDEX IF NOT EXISTS idx_runtime_error_triage_correlation
+    ON runtime_error_triage (correlated_consumer_fingerprint);
+
+-- =============================================================================
+-- COMMENTS
+-- =============================================================================
+
+COMMENT ON TABLE runtime_error_triage IS
+    'Incident tracking for runtime error events (OMN-5514). '
+    'Each row tracks an error incident by fingerprint for graduated triage. '
+    'Supports cross-layer correlation with consumer_health_triage via '
+    'correlated_consumer_fingerprint.';
+
+COMMENT ON COLUMN runtime_error_triage.fingerprint IS
+    'Stable hash of (logger_name, message_template, error_category) for deduplication.';
+
+COMMENT ON COLUMN runtime_error_triage.incident_state IS
+    'Lifecycle state: open, acknowledged, suppressed, ticketed, resolved.';
+
+COMMENT ON COLUMN runtime_error_triage.message_template IS
+    'Templatized error message with variable parts replaced by placeholders.';
+
+COMMENT ON COLUMN runtime_error_triage.correlated_consumer_fingerprint IS
+    'Links to consumer_health_triage.fingerprint for cross-layer correlation. '
+    'NULL if no consumer health correlation exists.';
+
+-- Update migration sentinel
+UPDATE public.db_metadata
+SET migrations_complete = TRUE,
+    schema_version = '055',
+    updated_at = NOW()
+WHERE id = TRUE AND owner_service = 'omnibase_infra';

docker/migrations/schema_fingerprint.sha256

@@ -1,6 +1,6 @@
 # Schema migration fingerprint for omnibase_infra (auto-generated)
 # Regenerate: python scripts/check_schema_fingerprint.py stamp
 # Verify:     python scripts/check_schema_fingerprint.py verify
-sha256:38841cefe359f6b19d8b39de275e9ef93b1842ef4a5a983d61ea7e2fcab11482
-generated_at: 2026-03-17T20:46:21Z
-migration_file_count: 32
+sha256:8974bc651d3282eb8ebb174ce224f653917600d543b60cb6c97b829f888eb30b
+generated_at: 2026-03-19T20:54:52Z
+migration_file_count: 34

scripts/check_contract_topic_parity.py

@@ -318,6 +318,10 @@
     "onex.evt.platform.resolution-decided.v1": "pre-migration; needs contract.yaml in omnibase_infra | owner: jonah | expiry: 2026-06-01",
     "onex.evt.platform.service-heartbeat.v1": "new topic added in OMN-5184; contract.yaml needed once consuming node is wired | owner: jonah | expiry: 2026-06-01",
     "onex.snapshot.platform.registration-snapshots.v1": "non-standard kind 'snapshot'; topic validated separately via ValidateTopicSuffix skip-list | owner: jonah | expiry: 2026-06-01",
+    # --- consumer health pipeline topics (OMN-5529) ---
+    "onex.evt.omnibase-infra.consumer-health.v1": "OMN-5515; contract.yaml will be added with NodeConsumerHealthTriageEffect in OMN-5520 | owner: jonah | expiry: 2026-06-01",
+    "onex.cmd.omnibase-infra.consumer-restart.v1": "OMN-5515; contract.yaml will be added with NodeConsumerHealthTriageEffect in OMN-5520 | owner: jonah | expiry: 2026-06-01",
+    "onex.evt.omnibase-infra.runtime-error.v1": "OMN-5517; contract.yaml will be added with NodeRuntimeErrorTriageEffect in OMN-5522 | owner: jonah | expiry: 2026-06-01",
 }
 # fmt: on
 

src/omnibase_infra/enums/generated/enum_omnibase_infra_topic.py

@@ -18,12 +18,15 @@ class EnumOmnibaseInfraTopic(str, Enum):
     All values are raw topic strings as declared in contract.yaml.
     Members are sorted by (kind, event_name, version).
     """
+    CMD_CONSUMER_RESTART_V1 = "onex.cmd.omnibase-infra.consumer-restart.v1"  # onex.cmd.omnibase-infra.consumer-restart.v1
     EVT_BASELINES_COMPUTED_V1 = "onex.evt.omnibase-infra.baselines-computed.v1"  # onex.evt.omnibase-infra.baselines-computed.v1
     EVT_CIRCUIT_BREAKER_STATE_V1 = "onex.evt.omnibase-infra.circuit-breaker-state.v1"  # onex.evt.omnibase-infra.circuit-breaker-state.v1
+    EVT_CONSUMER_HEALTH_V1 = "onex.evt.omnibase-infra.consumer-health.v1"  # onex.evt.omnibase-infra.consumer-health.v1
     EVT_DB_ERROR_V1 = "onex.evt.omnibase-infra.db-error.v1"  # onex.evt.omnibase-infra.db-error.v1
     EVT_EFFECTIVENESS_DATA_CHANGED_V1 = "onex.evt.omnibase-infra.effectiveness-data-changed.v1"  # onex.evt.omnibase-infra.effectiveness-data-changed.v1
     EVT_GMAIL_INTENT_RECEIVED_V1 = "onex.evt.omnibase-infra.gmail-intent-received.v1"  # onex.evt.omnibase-infra.gmail-intent-received.v1
     EVT_LLM_CALL_COMPLETED_V1 = "onex.evt.omnibase-infra.llm-call-completed.v1"  # onex.evt.omnibase-infra.llm-call-completed.v1
     EVT_LLM_ENDPOINT_HEALTH_V1 = "onex.evt.omnibase-infra.llm-endpoint-health.v1"  # onex.evt.omnibase-infra.llm-endpoint-health.v1
+    EVT_RUNTIME_ERROR_V1 = "onex.evt.omnibase-infra.runtime-error.v1"  # onex.evt.omnibase-infra.runtime-error.v1
     EVT_SAVINGS_ESTIMATED_V1 = "onex.evt.omnibase-infra.savings-estimated.v1"  # onex.evt.omnibase-infra.savings-estimated.v1
     EVT_WIRING_HEALTH_SNAPSHOT_V1 = "onex.evt.omnibase-infra.wiring-health-snapshot.v1"  # onex.evt.omnibase-infra.wiring-health-snapshot.v1