feat: close core Avo workflow gaps

Author: Astoriel

apps/api/app/api/dlq.py

@@ -1,18 +1,30 @@
 from __future__ import annotations
 
-from uuid import UUID
+import json
+from datetime import datetime, timezone
+from uuid import UUID, uuid5
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, Request
+from pydantic import ValidationError as PydanticValidationError
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.database import get_db
+from app.core.exceptions import BadRequestError
 from app.core.permissions import Permission, require_plan_permission
-from app.models import InvalidPayloadError
-from app.schemas.tracking_plan import InvalidPayloadErrorResponse
+from app.models import InvalidPayloadError, ValidationLog
+from app.schemas.tracking_plan import (
+    GuardDLQImportRecord,
+    GuardDLQImportResponse,
+    InvalidPayloadErrorResponse,
+)
+from app.services.snapshot_service import SnapshotService
 
 router = APIRouter(tags=["dlq"])
 
+GUARD_DLQ_IMPORT_NAMESPACE = UUID("f92c4282-7b14-4e2e-9ac1-0e6cf15f4a6c")
+MAX_IMPORT_RECORDS = 1000
+
 
 @router.get("/plans/{plan_id}/dlq", response_model=list[InvalidPayloadErrorResponse])
 async def get_dlq_errors(
@@ -27,3 +39,208 @@ async def get_dlq_errors(
         .limit(200)
     )
     return list(result.scalars().all())
+
+
+@router.post("/plans/{plan_id}/dlq/import", response_model=GuardDLQImportResponse)
+async def import_guard_dlq_errors(
+    plan_id: UUID,
+    request: Request,
+    access=Depends(require_plan_permission(Permission.EDIT)),
+    db: AsyncSession = Depends(get_db),
+):
+    del access
+    records = await _parse_import_records(request)
+    latest_version = await SnapshotService(db).get_latest_version(plan_id)
+    version_id = latest_version.id if latest_version else None
+
+    imported = 0
+    skipped = 0
+    upserted_errors = 0
+    for record in records:
+        request_id = uuid5(GUARD_DLQ_IMPORT_NAMESPACE, f"{plan_id}:{record.guard_dlq_id}")
+        existing_log = await db.execute(
+            select(ValidationLog.id).where(
+                ValidationLog.plan_id == plan_id,
+                ValidationLog.request_id == request_id,
+            )
+        )
+        if existing_log.scalar_one_or_none() is not None:
+            skipped += 1
+            continue
+
+        seen_at = _record_seen_at(record)
+        validation_log = ValidationLog(
+            plan_id=plan_id,
+            event_name=record.event_name,
+            payload=record.payload,
+            is_valid=False,
+            errors=_record_errors(record),
+            version_id=version_id,
+            api_key_id=None,
+            request_id=request_id,
+            source_ip=None,
+            source_label="guard-dlq-import",
+            validated_at=seen_at,
+        )
+        db.add(validation_log)
+        await db.flush()
+
+        await _upsert_imported_invalid_payload(
+            db,
+            plan_id=plan_id,
+            version_id=version_id,
+            validation_log_id=validation_log.id,
+            event_name=record.event_name,
+            payload=record.payload,
+            error_reason=_error_reason(record),
+            seen_at=seen_at,
+        )
+        imported += 1
+        upserted_errors += 1
+
+    return GuardDLQImportResponse(
+        imported=imported,
+        skipped=skipped,
+        upserted_errors=upserted_errors,
+    )
+
+
+async def _parse_import_records(request: Request) -> list[GuardDLQImportRecord]:
+    body = (await request.body()).strip()
+    if not body:
+        raise BadRequestError("Import body is empty.", code="dlq_import_empty")
+
+    try:
+        decoded = json.loads(body)
+    except json.JSONDecodeError:
+        decoded = _decode_ndjson(body)
+
+    if isinstance(decoded, dict) and "records" in decoded:
+        raw_records = decoded["records"]
+    elif isinstance(decoded, dict):
+        raw_records = [decoded]
+    else:
+        raw_records = decoded
+
+    if not isinstance(raw_records, list):
+        raise BadRequestError(
+            "DLQ import body must be NDJSON, a JSON array, or an object with records.",
+            code="dlq_import_invalid",
+        )
+    if not raw_records:
+        raise BadRequestError("DLQ import contains no records.", code="dlq_import_empty")
+    if len(raw_records) > MAX_IMPORT_RECORDS:
+        raise BadRequestError(
+            f"DLQ import accepts at most {MAX_IMPORT_RECORDS} records per request.",
+            code="dlq_import_too_large",
+        )
+
+    records: list[GuardDLQImportRecord] = []
+    for index, raw_record in enumerate(raw_records):
+        try:
+            records.append(GuardDLQImportRecord.model_validate(raw_record))
+        except PydanticValidationError as exc:
+            raise BadRequestError(
+                "DLQ import record is invalid.",
+                code="dlq_import_invalid_record",
+                extra={"record_index": index, "errors": exc.errors()},
+            ) from exc
+    return records
+
+
+def _decode_ndjson(body: bytes) -> list[dict]:
+    records: list[dict] = []
+    for line_number, raw_line in enumerate(body.splitlines(), start=1):
+        line = raw_line.strip()
+        if not line:
+            continue
+        try:
+            decoded = json.loads(line)
+        except json.JSONDecodeError as exc:
+            raise BadRequestError(
+                "DLQ import NDJSON line is invalid JSON.",
+                code="dlq_import_invalid_ndjson",
+                extra={"line": line_number},
+            ) from exc
+        if not isinstance(decoded, dict):
+            raise BadRequestError(
+                "DLQ import NDJSON lines must be JSON objects.",
+                code="dlq_import_invalid_ndjson",
+                extra={"line": line_number},
+            )
+        records.append(decoded)
+    return records
+
+
+def _record_seen_at(record: GuardDLQImportRecord) -> datetime:
+    seen_at = record.created_at or datetime.now(timezone.utc)
+    if seen_at.tzinfo is None:
+        return seen_at.replace(tzinfo=timezone.utc)
+    return seen_at.astimezone(timezone.utc)
+
+
+def _record_errors(record: GuardDLQImportRecord) -> list[dict]:
+    reason_codes = record.reason_codes or ["guard_dlq_reject"]
+    return [
+        {
+            "code": code,
+            "path": "payload",
+            "property_name": None,
+            "message": f"Guard rejected payload: {code}",
+        }
+        for code in reason_codes
+    ]
+
+
+def _error_reason(record: GuardDLQImportRecord) -> str:
+    reason_codes = ", ".join(record.reason_codes or ["guard_dlq_reject"])
+    return f"guard dlq rejected payload: {reason_codes}"
+
+
+async def _upsert_imported_invalid_payload(
+    db: AsyncSession,
+    *,
+    plan_id: UUID,
+    version_id: UUID | None,
+    validation_log_id: UUID,
+    event_name: str,
+    payload: dict,
+    error_reason: str,
+    seen_at: datetime,
+) -> None:
+    version_clause = (
+        InvalidPayloadError.version_id == version_id
+        if version_id is not None
+        else InvalidPayloadError.version_id.is_(None)
+    )
+    result = await db.execute(
+        select(InvalidPayloadError).where(
+            InvalidPayloadError.plan_id == plan_id,
+            version_clause,
+            InvalidPayloadError.event_name == event_name,
+            InvalidPayloadError.error_reason == error_reason,
+        )
+    )
+    invalid_payload = result.scalar_one_or_none()
+    if invalid_payload is None:
+        db.add(
+            InvalidPayloadError(
+                plan_id=plan_id,
+                version_id=version_id,
+                validation_log_id=validation_log_id,
+                event_name=event_name,
+                payload=payload,
+                error_reason=error_reason,
+                first_seen_at=seen_at,
+                last_seen_at=seen_at,
+                occurrence_count=1,
+            )
+        )
+        return
+
+    invalid_payload.payload = payload
+    invalid_payload.validation_log_id = validation_log_id
+    invalid_payload.last_seen_at = max(invalid_payload.last_seen_at or seen_at, seen_at)
+    if invalid_payload.first_seen_at is None or seen_at < invalid_payload.first_seen_at:
+        invalid_payload.first_seen_at = seen_at
+    invalid_payload.occurrence_count += 1

apps/api/app/api/implementation_status.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from uuid import UUID
+
+from fastapi import APIRouter, Depends
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.core.permissions import Permission, require_plan_permission
+from app.schemas.tracking_plan import ImplementationStatusResponse
+from app.services.implementation_status import ImplementationStatusService
+
+router = APIRouter(tags=["implementation-status"])
+
+
+@router.get(
+    "/plans/{plan_id}/implementation-status",
+    response_model=ImplementationStatusResponse,
+)
+async def get_implementation_status(
+    plan_id: UUID,
+    period: str = "all",
+    access=Depends(require_plan_permission(Permission.VIEW)),
+    db: AsyncSession = Depends(get_db),
+):
+    return await ImplementationStatusService(db).get_status(plan_id, period=period)

apps/api/app/api/router.py

@@ -8,6 +8,7 @@
 from app.api.dlq import router as dlq_router
 from app.api.dlq_triage import router as dlq_triage_router
 from app.api.health import router as health_router
+from app.api.implementation_status import router as implementation_status_router
 from app.api.merge_requests import router as merge_requests_router
 from app.api.org_settings import router as org_settings_router
 from app.api.plans import router as plans_router
@@ -19,6 +20,7 @@
 api_router.include_router(auth_router)
 api_router.include_router(plans_router)
 api_router.include_router(validation_router)
+api_router.include_router(implementation_status_router)
 api_router.include_router(versions_router)
 api_router.include_router(codegen_router)
 api_router.include_router(ai_router)

apps/api/app/schemas/tracking_plan.py

@@ -284,6 +284,28 @@ class ComplianceStats(BaseModel):
     period: str
 
 
+ImplementationEventStatus = Literal["never_seen", "seen_valid", "seen_invalid", "mixed"]
+
+
+class ImplementationStatusEvent(BaseModel):
+    event_name: str
+    status: ImplementationEventStatus
+    valid_count: int
+    invalid_count: int
+    last_seen_at: datetime | None = None
+    last_valid_at: datetime | None = None
+    last_invalid_at: datetime | None = None
+    source_labels: list[str] = Field(default_factory=list)
+    version_ids: list[uuid.UUID] = Field(default_factory=list)
+
+
+class ImplementationStatusResponse(BaseModel):
+    plan_id: uuid.UUID
+    period: str
+    generated_at: datetime
+    events: list[ImplementationStatusEvent] = Field(default_factory=list)
+
+
 class PublishPlanRequest(RevisionedRequest):
     summary: str | None = None
     allow_breaking: bool = False
@@ -357,6 +379,25 @@ class InvalidPayloadErrorResponse(BaseModel):
     model_config = {"from_attributes": True}
 
 
+class GuardDLQImportRecord(BaseModel):
+    schema_version: Literal["trackboard.guard.dlq.export.v1"]
+    source: str = Field(default="trackboard-guard", max_length=100)
+    guard_dlq_id: str = Field(min_length=1, max_length=300)
+    idempotency_key: str = Field(min_length=1, max_length=300)
+    event_name: str = Field(min_length=1, max_length=200)
+    severity: str = Field(default="blocked", max_length=40)
+    reason_codes: list[str] = Field(default_factory=list)
+    payload: dict[str, Any]
+    replay_status: str = Field(default="pending", max_length=40)
+    created_at: datetime | None = None
+
+
+class GuardDLQImportResponse(BaseModel):
+    imported: int
+    skipped: int
+    upserted_errors: int
+
+
 class DLQViolationSummary(BaseModel):
     code: str
     path: str
@@ -397,7 +438,7 @@ class DLQRecommendedActionResponse(BaseModel):
 
 class DLQTriageResponse(BaseModel):
     fingerprint: str
-    status: Literal["ready"]
+    status: Literal["ready", "deterministic_only"]
     summary: str
     confidence: str
     evidence: list[str] = Field(default_factory=list)