wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

feat(awooop): apply source correlation links
All checks were successful
Code Review / ai-code-review (push) Successful in 12s
Details
CD Pipeline / tests (push) Successful in 4m1s
Details
CD Pipeline / build-and-deploy (push) Successful in 4m8s
Details
CD Pipeline / post-deploy-checks (push) Successful in 2m2s
Details

Browse Source
This commit is contained in:
Your Name
2026-05-21 10:23:29 +08:00
parent d25237a31f
commit fe3bf5dc18
8 changed files with 1011 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

493
apps/api/src/services/channel_event_dossier_service.py
View File

@@ -29,7 +29,10 @@ _SOURCE_CORRELATION_WORK_ITEM_ID_MAX = 180
_SOURCE_CORRELATION_DECISION_SCHEMA_VERSION = (
"awooop_source_correlation_review_decision_v1"
)
_SOURCE_CORRELATION_APPLY_SCHEMA_VERSION = "awooop_source_correlation_apply_v1"
_SOURCE_CORRELATION_APPLY_STAGE = "source_correlation_linked"
_SOURCE_CORRELATION_REVIEW_ACTOR = "awooop_source_correlation_review_service"
_SOURCE_CORRELATION_APPLY_ACTOR = "awooop_source_correlation_apply_service"
_INCIDENT_ID_RE = re.compile(r"\bINC-\d{8}-[A-Z0-9]{4,}\b")
RecurrenceWorkItemMode = Literal["auto", "ticket", "reverify", "approval_review", "observe"]
RecurrenceWorkItemHandoffKind = Literal["ticket_proposal", "manual_review"]
@@ -155,6 +158,43 @@ def _normalize_source_review_decision(row: dict[str, Any]) -> dict[str, Any] | N
}
def _normalize_source_apply(row: dict[str, Any]) -> dict[str, Any] | None:
context = _as_dict(row.get("context"))
if context.get("schema_version") != _SOURCE_CORRELATION_APPLY_SCHEMA_VERSION:
return None
work_item_id = str(context.get("work_item_id") or "").strip()
target_incident_id = str(context.get("target_incident_id") or "").strip()
if not work_item_id or not target_incident_id:
return None
history = _as_dict(context.get("history"))
return {
"schema_version": _SOURCE_CORRELATION_APPLY_SCHEMA_VERSION,
"apply_id": row.get("id"),
"work_item_id": work_item_id,
"apply_status": context.get("apply_status") or "applied",
"target_incident_id": target_incident_id,
"review_id": context.get("review_id"),
"reviewer_id": context.get("reviewer_id"),
"operator_note": context.get("operator_note"),
"latest_provider_event_id": context.get("latest_provider_event_id"),
"source_event_provider_event_id": context.get(
"source_event_provider_event_id"
),
"source_event_id": history.get("source_event_id"),
"timeline_event_id": history.get("timeline_event_id"),
"recorded_at": row.get("created_at"),
"history": history,
}
def _provider_raw_event_id(provider_event_id: Any) -> str:
parts = str(provider_event_id or "").split(":", 2)
if len(parts) == 3 and parts[2].strip():
return parts[2].strip()
return str(provider_event_id or "unknown").strip() or "unknown"
def _append_unique(values: list[str], candidate: Any) -> None:
text_value = str(candidate or "").strip()
if text_value and text_value not in values:
@@ -210,11 +250,13 @@ def build_dossier_recurrence(
limit: int,
repair_summaries_by_incident: dict[str, dict[str, Any]] | None = None,
source_review_decisions_by_work_item: dict[str, dict[str, Any]] | None = None,
source_applies_by_work_item: dict[str, dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Group recent source events into recurrence buckets with linked run state."""
groups: dict[str, dict[str, Any]] = {}
repair_summaries = repair_summaries_by_incident or {}
source_review_decisions = source_review_decisions_by_work_item or {}
source_applies = source_applies_by_work_item or {}
for row in rows:
event = build_dossier_event(row)
@@ -258,6 +300,7 @@ def build_dossier_recurrence(
"first_received_at": received_at,
"latest_received_at": received_at,
"_run_ids": set(),
"_source_correlation_work_item_ids": [],
},
)
@@ -271,6 +314,20 @@ def build_dossier_recurrence(
stage = str(event.get("stage") or "received")
stage_counts = group["stage_counts"]
stage_counts[stage] = int(stage_counts.get(stage, 0)) + 1
if (
str(event.get("provider") or "").lower()
in _SOURCE_CORRELATION_REVIEW_PROVIDERS
and stage.lower() not in _SOURCE_CORRELATION_REVIEW_EXCLUDED_STAGES
and source_ref_count > 0
and not incident_ids
):
_append_unique(
group["_source_correlation_work_item_ids"],
_source_correlation_work_item_id({
"latest_provider_event_id": event.get("provider_event_id"),
"recurrence_key": key,
}),
)
for incident_id in incident_ids:
_append_unique(group["incident_ids"], incident_id)
@@ -316,6 +373,7 @@ def build_dossier_recurrence(
group,
repair_summaries,
source_review_decisions,
source_applies,
)
linked_run_total += len(run_ids)
items.append(group)
@@ -387,6 +445,9 @@ def build_dossier_recurrence(
"source_correlation_decision_recorded_group_total": sum(
1 for item in items if item.get("source_correlation_review")
),
"source_correlation_applied_group_total": sum(
1 for item in items if item.get("source_correlation_apply")
),
"latest_received_at": latest_received_at,
},
"items": items,
@@ -457,6 +518,7 @@ def _work_item_next_step(repair_status: str) -> str:
"source_correlation_review": "review_provider_source_match",
"source_correlation_accepted": "verify_source_match_in_status_chain",
"source_correlation_rejected": "monitor_for_new_provider_evidence",
"source_correlation_applied": "verify_source_link_in_status_chain",
"auto_repair_succeeded_unverified": "run_post_verification",
"auto_repair_failed": "triage_failed_repair",
"auto_repair_recorded": "review_repair_record",
@@ -472,6 +534,7 @@ def _work_item_reason(repair_status: str) -> str:
"source_correlation_review": "provider_native_evidence_unlinked",
"source_correlation_accepted": "provider_native_evidence_accepted",
"source_correlation_rejected": "provider_native_evidence_rejected",
"source_correlation_applied": "provider_native_evidence_link_applied",
"auto_repair_succeeded_unverified": "auto_repair_missing_verification",
"auto_repair_failed": "auto_repair_failed",
"auto_repair_recorded": "auto_repair_record_needs_review",
@@ -486,6 +549,7 @@ def _attach_work_item_summary(
group: dict[str, Any],
repair_summaries_by_incident: dict[str, dict[str, Any]],
source_review_decisions_by_work_item: dict[str, dict[str, Any]],
source_applies_by_work_item: dict[str, dict[str, Any]],
) -> None:
incident_ids = [
str(incident_id) for incident_id in group.get("incident_ids", []) if incident_id
@@ -533,10 +597,31 @@ def _attach_work_item_summary(
elif status_value == "source_correlation_review" and work_status != "none":
work_item_id = _source_correlation_work_item_id(group)
source_review_decision = (
source_review_decisions_by_work_item.get(work_item_id)
if work_item_id
else None
source_work_item_ids = [
str(candidate)
for candidate in group.pop("_source_correlation_work_item_ids", [])
if candidate
]
candidate_work_item_ids = [
candidate
for candidate in [work_item_id, *source_work_item_ids]
if candidate
]
source_review_decision = next(
(
source_review_decisions_by_work_item[candidate]
for candidate in candidate_work_item_ids
if candidate in source_review_decisions_by_work_item
),
None,
)
source_apply = next(
(
source_applies_by_work_item[candidate]
for candidate in candidate_work_item_ids
if candidate in source_applies_by_work_item
),
None,
)
matched_incident_id = None
work_item_next_step = _work_item_next_step(status_value)
@@ -552,6 +637,15 @@ def _attach_work_item_summary(
work_item_next_step = outcome["next_step"]
work_item_reason = outcome["reason"]
group["source_correlation_review"] = source_review_decision
if source_apply:
matched_incident_id = source_apply.get("target_incident_id")
group["source_correlation_apply"] = source_apply
if status_value in {
"source_correlation_review",
"source_correlation_accepted",
}:
work_item_next_step = "verify_source_link_in_status_chain"
work_item_reason = "provider_native_evidence_link_applied"
group["latest_incident_id"] = latest_incident_id
group["repair_summary"] = repair_payload
@@ -762,6 +856,7 @@ def _recurrence_current_state_summary(
"latest_run_id": item.get("latest_run_id"),
"matched_incident_id": work_item.get("matched_incident_id"),
"source_correlation_review": item.get("source_correlation_review"),
"source_correlation_apply": item.get("source_correlation_apply"),
"repair_status": repair_summary.get("status"),
"latest_auto_repair_id": repair_summary.get("latest_auto_repair_id"),
"latest_verification_result": repair_summary.get("latest_verification_result"),
@@ -1409,6 +1504,314 @@ async def _record_source_correlation_review_decision_history(
return history
def _source_correlation_apply_checks(
item: dict[str, Any],
work_item: dict[str, Any],
source_review: dict[str, Any],
*,
target_incident_id: str | None,
) -> list[dict[str, Any]]:
provider = str(item.get("provider") or "").lower()
source_ref_total = int(item.get("source_ref_total") or 0)
return [
{
"name": "source_review_work_item",
"passed": work_item.get("kind") == "source_correlation_review",
"detail": str(work_item.get("kind") or "unknown"),
},
{
"name": "accepted_review_recorded",
"passed": source_review.get("decision") == "accepted",
"detail": str(source_review.get("decision") or "missing"),
},
{
"name": "target_incident_present",
"passed": bool(target_incident_id),
"detail": target_incident_id or "missing target_incident_id",
},
{
"name": "provider_supported",
"passed": provider in _SOURCE_CORRELATION_REVIEW_PROVIDERS,
"detail": provider or "unknown",
},
{
"name": "source_refs_present",
"passed": source_ref_total > 0,
"detail": str(source_ref_total),
},
{
"name": "provider_event_present",
"passed": bool(item.get("latest_provider_event_id")),
"detail": str(item.get("latest_provider_event_id") or "missing"),
},
{
"name": "append_only_source_link",
"passed": True,
"detail": "awooop_conversation_event_append_only",
},
]
def _source_correlation_apply_context(payload: dict[str, Any]) -> dict[str, Any]:
return {
"schema_version": _SOURCE_CORRELATION_APPLY_SCHEMA_VERSION,
"source": payload.get("source"),
"project_id": payload.get("project_id"),
"work_item_id": payload.get("work_item_id"),
"review_id": payload.get("review_id"),
"apply_status": payload.get("apply_status"),
"target_incident_id": payload.get("target_incident_id"),
"reviewer_id": payload.get("reviewer_id"),
"operator_note": payload.get("operator_note"),
"latest_provider_event_id": payload.get("latest_provider_event_id"),
"source_event_provider_event_id": payload.get(
"source_event_provider_event_id"
),
"source_event_stage": payload.get("source_event_stage"),
"provider": payload.get("provider"),
"alertname": payload.get("alertname"),
"namespace": payload.get("namespace"),
"target_resource": payload.get("target_resource"),
"safety_level": payload.get("safety_level"),
"writes_incident_state": payload.get("writes_incident_state"),
"writes_source_event": payload.get("writes_source_event"),
"writes_auto_repair_result": payload.get("writes_auto_repair_result"),
"writes_ticket": payload.get("writes_ticket"),
"creates_external_ticket": payload.get("creates_external_ticket"),
"checks": payload.get("checks"),
"current_state_summary": payload.get("current_state_summary"),
"plan": payload.get("plan"),
"read_model_route": payload.get("read_model_route"),
"next_step": payload.get("next_step"),
"history": payload.get("history"),
}
def _source_correlation_apply_description(context: dict[str, Any]) -> str:
return (
f"work_item={context.get('work_item_id')} "
f"provider_event={context.get('latest_provider_event_id')} "
f"source_event={context.get('source_event_provider_event_id')} "
f"target_incident={context.get('target_incident_id')} "
f"writes_source_event={context.get('writes_source_event')} "
f"writes_incident={context.get('writes_incident_state')}"
)[:500]
def build_source_correlation_apply(
recurrence: dict[str, Any],
*,
work_item_id: str,
reviewer_id: str = "operator_console",
operator_note: str | None = None,
) -> dict[str, Any]:
"""Build an append-only source link apply payload for an accepted review."""
item, work_item = _find_recurrence_work_item(recurrence, work_item_id)
source_review = _as_dict(item.get("source_correlation_review"))
target_id = (
str(
source_review.get("target_incident_id")
or work_item.get("matched_incident_id")
or ""
).strip()
or None
)
checks = _source_correlation_apply_checks(
item,
work_item,
source_review,
target_incident_id=target_id,
)
allowed = all(check["passed"] for check in checks)
raw_event_id = _provider_raw_event_id(item.get("latest_provider_event_id"))
provider = str(item.get("provider") or "external").strip().lower() or "external"
source_event_provider_event_id = (
f"{provider}:{_SOURCE_CORRELATION_APPLY_STAGE}:{raw_event_id[:96]}"
)
payload = {
"schema_version": _SOURCE_CORRELATION_APPLY_SCHEMA_VERSION,
"source": "channel_event_dossier.recurrence",
"project_id": recurrence.get("project_id"),
"work_item_id": work_item.get("work_item_id"),
"review_id": source_review.get("review_id"),
"target_incident_id": target_id,
"reviewer_id": str(reviewer_id or "operator_console")[:100],
"operator_note": str(operator_note or "").strip()[:500] or None,
"allowed": allowed,
"executed": allowed,
"apply_status": "ready_to_apply" if allowed else "blocked",
"safety_level": "source_review_append_only_apply",
"writes_incident_state": False,
"writes_source_event": allowed,
"writes_auto_repair_result": False,
"writes_ticket": False,
"creates_external_ticket": False,
"latest_provider_event_id": item.get("latest_provider_event_id"),
"source_event_provider_event_id": source_event_provider_event_id,
"source_event_stage": _SOURCE_CORRELATION_APPLY_STAGE,
"raw_event_id": raw_event_id,
"provider": provider,
"alertname": item.get("alertname"),
"severity": item.get("severity"),
"namespace": item.get("namespace"),
"target_resource": item.get("target_resource"),
"fingerprint": item.get("fingerprint"),
"checks": checks,
"current_state_summary": _recurrence_current_state_summary(item, work_item),
"plan": {
"step": "append_source_correlation_link_event",
"flywheel_node": "source_correlation_apply",
"agent_id": "awooop_source_correlation_reviewer",
"required_scope": "append_source_link",
"writes": (
["awooop_conversation_event", "timeline_events", "alert_operation_log"]
if allowed
else []
),
"target_action": "verify_source_link_in_status_chain",
"reason": (
"provider_native_evidence_link_applied"
if allowed
else "source_correlation_apply_preflight_failed"
),
"target": _recurrence_work_item_target(item),
},
"read_model_route": {
"agent_id": "awooop_source_correlation_reviewer",
"tool_name": "channel_event_dossier.recurrence",
"required_scope": "append_source_link",
"is_shadow": False,
"flywheel_node": "source_correlation_apply",
},
"next_step": (
"verify_source_link_in_status_chain"
if allowed
else "fix_preflight_checks"
),
}
if not allowed:
payload["executed"] = False
return payload
async def _record_source_correlation_apply_history(
payload: dict[str, Any],
) -> dict[str, Any]:
if not payload.get("allowed"):
return {"recorded": False, "reason": "preflight_failed"}
history: dict[str, Any] = {
"recorded": False,
"source_event_id": None,
"alert_operation_id": None,
"timeline_event_id": None,
}
incident_id = str(payload.get("target_incident_id") or "")
try:
from src.services.channel_hub import record_external_alert_event
event_id = await record_external_alert_event(
project_id=str(payload.get("project_id") or "awoooi"),
provider=str(payload.get("provider") or "external"),
event_id=str(payload.get("raw_event_id") or payload.get("work_item_id")),
stage=_SOURCE_CORRELATION_APPLY_STAGE,
title=str(payload.get("alertname") or "Source correlation link"),
severity=str(payload.get("severity") or "info"),
namespace=payload.get("namespace"),
target_resource=payload.get("target_resource"),
fingerprint=payload.get("fingerprint"),
incident_id=incident_id,
source_url=None,
labels={
"awooop_source_correlation": "applied",
"work_item_id": payload.get("work_item_id"),
},
annotations={
"review_id": payload.get("review_id") or "",
"original_provider_event_id": payload.get("latest_provider_event_id")
or "",
},
payload={
"schema_version": _SOURCE_CORRELATION_APPLY_SCHEMA_VERSION,
"work_item_id": payload.get("work_item_id"),
"review_id": payload.get("review_id"),
"target_incident_id": incident_id,
},
is_duplicate=False,
)
if event_id:
history["source_event_id"] = str(event_id)
except Exception as exc:
logger.warning(
"awooop_source_correlation_apply_source_event_failed",
work_item_id=payload.get("work_item_id"),
incident_id=incident_id,
error=str(exc),
)
context = _source_correlation_apply_context({**payload, "history": history})
try:
from src.services.approval_db import get_timeline_service
event = await get_timeline_service().add_event(
event_type="human",
status="success" if history.get("source_event_id") else "warning",
title="AwoooP source correlation apply",
description=_source_correlation_apply_description(context),
actor=_SOURCE_CORRELATION_APPLY_ACTOR,
actor_role="source_correlation_apply",
incident_id=incident_id,
)
if event:
history["timeline_event_id"] = event.get("id")
except Exception as exc:
logger.warning(
"awooop_source_correlation_apply_timeline_failed",
incident_id=incident_id,
work_item_id=payload.get("work_item_id"),
error=str(exc),
)
try:
from src.repositories.alert_operation_log_repository import (
get_alert_operation_log_repository,
)
final_context = _source_correlation_apply_context({
**payload,
"apply_status": "applied" if history.get("source_event_id") else "partial",
"history": history,
})
record = await get_alert_operation_log_repository().append(
"USER_ACTION",
incident_id=incident_id,
actor=_SOURCE_CORRELATION_APPLY_ACTOR,
action_detail="source_correlation_apply:append_source_link",
success=bool(history.get("source_event_id")),
context=_json_safe(final_context),
)
if record is not None:
history["alert_operation_id"] = getattr(record, "id", None)
except Exception as exc:
logger.warning(
"awooop_source_correlation_apply_alert_operation_failed",
work_item_id=payload.get("work_item_id"),
error=str(exc),
)
history["recorded"] = bool(
history.get("source_event_id")
or history.get("alert_operation_id")
or history.get("timeline_event_id")
)
if not history["recorded"]:
history["reason"] = "history_sink_unavailable"
return history
def build_dossier_coverage(
rows: list[dict[str, Any]],
*,
@@ -1681,6 +2084,48 @@ async def _fetch_source_review_decisions_by_work_item(
return decisions
async def _fetch_source_applies_by_work_item(
db: Any,
*,
project_id: str,
limit: int,
) -> dict[str, dict[str, Any]]:
"""Fetch latest source-correlation apply records from event-sourced audit."""
result = await db.execute(
text("""
SELECT DISTINCT ON (context->>'work_item_id')
id,
incident_id,
actor,
action_detail,
success,
context,
created_at
FROM alert_operation_log
WHERE actor = :actor
AND context->>'schema_version' = :schema_version
AND COALESCE(context->>'project_id', :project_id) = :project_id
ORDER BY context->>'work_item_id', created_at DESC
LIMIT :limit
"""),
{
"actor": _SOURCE_CORRELATION_APPLY_ACTOR,
"schema_version": _SOURCE_CORRELATION_APPLY_SCHEMA_VERSION,
"project_id": project_id,
"limit": max(1, min(limit, _MAX_RECURRENCE_EVENTS)),
},
)
applies: dict[str, dict[str, Any]] = {}
for row in result.mappings().all():
item = _normalize_source_apply(dict(row))
if not item:
continue
applies[str(item["work_item_id"])] = item
return applies
async def fetch_channel_event_dossier(
*,
project_id: str | None,
@@ -1872,6 +2317,11 @@ async def fetch_channel_event_dossier_recurrence(
project_id=effective_project_id,
limit=safe_limit,
)
source_applies = await _fetch_source_applies_by_work_item(
db,
project_id=effective_project_id,
limit=safe_limit,
)
return build_dossier_recurrence(
rows,
@@ -1879,6 +2329,7 @@ async def fetch_channel_event_dossier_recurrence(
limit=safe_limit,
repair_summaries_by_incident=repair_summaries,
source_review_decisions_by_work_item=source_review_decisions,
source_applies_by_work_item=source_applies,
)
@@ -1994,3 +2445,37 @@ async def fetch_source_correlation_review_decision(
else:
payload["review_record_status"] = "blocked"
return payload
async def fetch_source_correlation_apply(
*,
project_id: str | None,
work_item_id: str,
reviewer_id: str = "operator_console",
operator_note: str | None = None,
provider: str | None = None,
limit: int = _MAX_RECURRENCE_EVENTS,
) -> dict[str, Any]:
"""Fetch recurrence state and append an accepted source review link event."""
recurrence = await fetch_channel_event_dossier_recurrence(
project_id=project_id,
provider=provider,
limit=limit,
)
payload = build_source_correlation_apply(
recurrence,
work_item_id=work_item_id,
reviewer_id=reviewer_id,
operator_note=operator_note,
)
payload["history"] = await _record_source_correlation_apply_history(payload)
if payload["history"].get("source_event_id"):
payload["apply_status"] = "applied"
elif payload["history"].get("recorded") and payload.get("allowed"):
payload["apply_status"] = "partial"
elif payload.get("allowed"):
payload["apply_status"] = "record_failed"
else:
payload["apply_status"] = "blocked"
return payload