Kairo-Jinkendo/backend/steering/hooks/dispatch.py
Lars 95feba6204
All checks were successful
Deploy Development / deploy (push) Successful in 44s
Test Suite / pytest-backend (push) Successful in 1m19s
Test Suite / lint-backend (push) Successful in 2s
Test Suite / compose-smoke (push) Has been skipped
Test Suite / k6 /api/health Baseline (push) Successful in 18s
Test Suite / playwright-smoke (push) Successful in 11s
AP1.0: Steering Foundation — Lifecycle, Context, Signal Engine
Persistierter Standard Lifecycle pro Initiative, backend/steering/ Skeleton, Attention-Refactor und Hook-Dispatch als Basis für Method Registry.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-05 17:09:20 +02:00

36 lines
975 B
Python

"""Hook dispatch — AP1.0 stub with audit trail."""
from __future__ import annotations
from typing import Any, Optional
from steering.hooks.registry import get_hook
from tenant_context import TenantContext
def dispatch_hook(
slug: str,
ctx: TenantContext | None,
payload: dict[str, Any],
*,
tenant_id: Optional[str] = None,
user_id: Optional[str] = None,
) -> list[dict[str, Any]]:
"""Dispatch hook — AP1.0: audit only, no handlers yet."""
hook = get_hook(slug)
if not hook:
return [{"kind": "hook_unknown", "slug": slug}]
tid = tenant_id or (ctx.tenant_id if ctx else None)
if tid and user_id:
from services.audit import log_audit
log_audit(
f"steering.hook.{slug}",
user_id=user_id,
tenant_id=tid,
details={"slug": slug, "payload": payload},
)
return [{"kind": "hook_dispatched", "slug": slug, "lifecycle_step": hook.lifecycle_step}]