16e791e044
8월 7일까지 워킹트리에만 남아 있던 미커밋 작업을 커밋한다. 여러 사본 폴더(worktree·clone)에 흩어져 있던 중간 스냅샷을 정리하기 전에 원본을 git 이력으로 고정하는 것이 목적이다. - contracts/routes/services: measurement, outcome_trajectory, rupture_repair, deliberate_practice, calibration_transfer, supervision_research, multimodal_alliance, continuous_improvement 계열 신규 모듈과 테스트 - infra/db/init: 07~16 마이그레이션(측정 기반~calibration transfer 실행) - apps/web: 세션 리뷰 카드·관리 화면·E2E 스펙 추가 - docs/ops: G0~G8 라이브 통합·배포·롤백 증거 문서와 evidence JSON/PNG - scripts: smoke·ledger·릴리스 에이전트·NAS 프리뷰 운영 스크립트 engine.public 로그 .bak과 apps/web/test-results 산출물은 커밋에서 제외했다.
511 lines
18 KiB
Python
511 lines
18 KiB
Python
"""Durable scheduled producer for the G8 agentic content pipeline.
|
|
|
|
The producer is intentionally opt-in. It only claims source packs already
|
|
classified as synthetic research material, revalidates source approval/hash/PII
|
|
before every engine run, and can only persist a pending human-review candidate.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
from uuid import UUID, uuid5
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator
|
|
|
|
from .. import db
|
|
from ..config import settings
|
|
from ..engine_client import EngineClient, engine_client
|
|
from . import continuous_improvement_agentic, continuous_improvement_store
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DATA_CLASSIFICATION = "synthetic_replay_red_team_coverage_drift"
|
|
_PRODUCER_NAMESPACE = UUID("87cd17b8-2e03-5ea9-9454-7ff60d7ba142")
|
|
_REPO_SOURCE_PATH = (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "data"
|
|
/ "continuous_improvement"
|
|
/ "synthetic_source_pack.v5.json"
|
|
)
|
|
_PRODUCER_TASK: asyncio.Task[None] | None = None
|
|
|
|
|
|
class AgenticJobError(ValueError):
|
|
"""Durable queue specification or transition failed."""
|
|
|
|
|
|
class ScheduledAgenticJobSpec(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
job_key: str = Field(pattern=r"^oas-g8-job-[a-z0-9-]+$")
|
|
data_classification: Literal["synthetic_replay_red_team_coverage_drift"]
|
|
content_kind: Literal["case", "rupture", "practice", "benchmark"]
|
|
difficulty_level: int = Field(ge=1, le=5)
|
|
variant_count: int = Field(ge=3, le=12)
|
|
prompt_version: str = Field(min_length=1, max_length=80)
|
|
trigger_kind: Literal["scheduled_repo_source", "scheduled_incident"]
|
|
source_packs: tuple[continuous_improvement_agentic.AgenticSourcePack, ...] = Field(
|
|
min_length=1,
|
|
max_length=20,
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_sources_before_enqueue(self) -> "ScheduledAgenticJobSpec":
|
|
continuous_improvement_agentic.validate_source_packs(self.source_packs)
|
|
return self
|
|
|
|
|
|
class ClaimedAgenticJob(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
job_id: UUID
|
|
spec: ScheduledAgenticJobSpec
|
|
source_fingerprint: str = Field(pattern=r"^[a-f0-9]{64}$")
|
|
attempt_count: int = Field(ge=1)
|
|
|
|
|
|
class AgenticJobOutcome(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
job_id: UUID
|
|
status: Literal["completed", "retry_wait", "rejected"]
|
|
agent_calls_executed: int = Field(ge=0)
|
|
idempotent_replay: bool = False
|
|
error_code: str | None = None
|
|
|
|
|
|
class _ProducerEngine:
|
|
"""Apply the G8 call timeout without changing other EngineClient users."""
|
|
|
|
def __init__(self, engine: Any) -> None:
|
|
self.engine = engine
|
|
|
|
async def generate(self, request: Any) -> Any:
|
|
if isinstance(self.engine, EngineClient):
|
|
return await self.engine.generate(
|
|
request,
|
|
timeout=settings.continuous_improvement_producer_engine_timeout_seconds,
|
|
)
|
|
return await self.engine.generate(request)
|
|
|
|
|
|
def _canonical_json(value: Any) -> str:
|
|
return json.dumps(
|
|
value,
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
default=str,
|
|
)
|
|
|
|
|
|
def _fingerprint(spec: ScheduledAgenticJobSpec) -> str:
|
|
return hashlib.sha256(
|
|
_canonical_json(spec.model_dump(mode="json")).encode("utf-8")
|
|
).hexdigest()
|
|
|
|
|
|
def _job_ids(job_id: UUID) -> dict[str, UUID]:
|
|
return {
|
|
name: uuid5(job_id, name)
|
|
for name in (
|
|
"submission",
|
|
"pipeline",
|
|
"benchmark_record",
|
|
"qualification",
|
|
)
|
|
}
|
|
|
|
|
|
def load_repo_approved_job(
|
|
path: Path = _REPO_SOURCE_PATH,
|
|
) -> ScheduledAgenticJobSpec:
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
return ScheduledAgenticJobSpec.model_validate(payload)
|
|
except (OSError, json.JSONDecodeError, ValidationError) as exc:
|
|
raise AgenticJobError("repo-approved G8 source pack is invalid") from exc
|
|
|
|
|
|
async def enqueue_agentic_job(conn: Any, spec: ScheduledAgenticJobSpec) -> UUID:
|
|
"""Idempotently enqueue an immutable validated source/configuration tuple."""
|
|
|
|
# Revalidate at the persistence boundary even when caller already has a model.
|
|
continuous_improvement_agentic.validate_source_packs(spec.source_packs)
|
|
if spec.data_classification != DATA_CLASSIFICATION:
|
|
raise AgenticJobError("unsupported agentic job data classification")
|
|
job_id = uuid5(_PRODUCER_NAMESPACE, spec.job_key)
|
|
source_fingerprint = _fingerprint(spec)
|
|
await conn.execute(
|
|
"""
|
|
INSERT INTO app.ci_agentic_job (
|
|
job_id, job_key, source_packs, source_fingerprint, data_classification,
|
|
content_kind, difficulty_level, variant_count, prompt_version, trigger_kind
|
|
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
|
|
ON CONFLICT (job_id) DO NOTHING
|
|
""",
|
|
job_id,
|
|
spec.job_key,
|
|
[item.model_dump(mode="json") for item in spec.source_packs],
|
|
source_fingerprint,
|
|
spec.data_classification,
|
|
spec.content_kind,
|
|
spec.difficulty_level,
|
|
spec.variant_count,
|
|
spec.prompt_version,
|
|
spec.trigger_kind,
|
|
)
|
|
row = await conn.fetchrow(
|
|
"""
|
|
SELECT job_key, source_fingerprint, data_classification
|
|
FROM app.ci_agentic_job WHERE job_id = $1
|
|
""",
|
|
job_id,
|
|
)
|
|
if row is None:
|
|
raise AgenticJobError("agentic job is not visible after enqueue")
|
|
if (
|
|
str(row["job_key"]) != spec.job_key
|
|
or str(row["source_fingerprint"]) != source_fingerprint
|
|
or str(row["data_classification"]) != DATA_CLASSIFICATION
|
|
):
|
|
raise AgenticJobError("agentic job key was reused with changed source input")
|
|
return job_id
|
|
|
|
|
|
async def ensure_repo_approved_job() -> UUID:
|
|
spec = load_repo_approved_job()
|
|
async with db.acquire(ai_view="research", ai_context=True) as conn:
|
|
return await enqueue_agentic_job(conn, spec)
|
|
|
|
|
|
async def claim_next_agentic_job(
|
|
*, job_id: UUID | None = None
|
|
) -> ClaimedAgenticJob | None:
|
|
async with db.acquire(ai_view="research", ai_context=True) as conn:
|
|
row = await conn.fetchrow(
|
|
"""
|
|
WITH candidate AS (
|
|
SELECT job_id
|
|
FROM app.ci_agentic_job
|
|
WHERE (
|
|
(status IN ('pending','retry_wait') AND next_attempt_at <= now())
|
|
OR (
|
|
status = 'processing'
|
|
AND lease_started_at <= now() - ($1::double precision * interval '1 second')
|
|
)
|
|
)
|
|
AND ($2::uuid IS NULL OR job_id = $2)
|
|
ORDER BY next_attempt_at, created_at
|
|
FOR UPDATE SKIP LOCKED
|
|
LIMIT 1
|
|
)
|
|
UPDATE app.ci_agentic_job job
|
|
SET status = 'processing',
|
|
attempt_count = job.attempt_count + 1,
|
|
lease_started_at = now(),
|
|
updated_at = now(),
|
|
last_error_code = NULL,
|
|
last_error_message = NULL
|
|
FROM candidate
|
|
WHERE job.job_id = candidate.job_id
|
|
RETURNING job.*
|
|
""",
|
|
settings.continuous_improvement_producer_lease_timeout_seconds,
|
|
job_id,
|
|
)
|
|
if row is None:
|
|
return None
|
|
try:
|
|
spec = ScheduledAgenticJobSpec.model_validate(
|
|
{
|
|
"job_key": row["job_key"],
|
|
"data_classification": row["data_classification"],
|
|
"content_kind": row["content_kind"],
|
|
"difficulty_level": row["difficulty_level"],
|
|
"variant_count": row["variant_count"],
|
|
"prompt_version": row["prompt_version"],
|
|
"trigger_kind": row["trigger_kind"],
|
|
"source_packs": row["source_packs"],
|
|
}
|
|
)
|
|
return ClaimedAgenticJob(
|
|
job_id=row["job_id"],
|
|
spec=spec,
|
|
source_fingerprint=str(row["source_fingerprint"]),
|
|
attempt_count=int(row["attempt_count"]),
|
|
)
|
|
except (ValidationError, continuous_improvement_agentic.AgenticPipelineRejectedError) as exc:
|
|
# The row was durable before validation failed. Preserve a terminal,
|
|
# inspectable rejection without exposing source/model payloads in errors.
|
|
job_id = UUID(str(row["job_id"]))
|
|
await _mark_rejected(job_id, "invalid_source_contract", str(exc))
|
|
raise AgenticJobError("claimed job failed source validation") from exc
|
|
|
|
|
|
def _safe_error_message(exc: BaseException) -> str:
|
|
return f"{type(exc).__name__}: {str(exc)}"[:500]
|
|
|
|
|
|
async def _mark_retry(job_id: UUID, error_code: str, exc: BaseException) -> None:
|
|
async with db.acquire(ai_view="research", ai_context=True) as conn:
|
|
await conn.execute(
|
|
"""
|
|
UPDATE app.ci_agentic_job
|
|
SET status = 'retry_wait', lease_started_at = NULL,
|
|
next_attempt_at = now() + ($2::double precision * interval '1 second'),
|
|
last_error_code = $3, last_error_message = $4, updated_at = now()
|
|
WHERE job_id = $1 AND status = 'processing'
|
|
""",
|
|
job_id,
|
|
settings.continuous_improvement_producer_retry_delay_seconds,
|
|
error_code,
|
|
_safe_error_message(exc),
|
|
)
|
|
|
|
|
|
async def _mark_rejected(job_id: UUID, error_code: str, message: str) -> None:
|
|
async with db.acquire(ai_view="research", ai_context=True) as conn:
|
|
await conn.execute(
|
|
"""
|
|
UPDATE app.ci_agentic_job
|
|
SET status = 'rejected', lease_started_at = NULL,
|
|
last_error_code = $2, last_error_message = $3, updated_at = now()
|
|
WHERE job_id = $1 AND status = 'processing'
|
|
""",
|
|
job_id,
|
|
error_code,
|
|
message[:500],
|
|
)
|
|
|
|
|
|
async def execute_claimed_agentic_job(
|
|
job: ClaimedAgenticJob,
|
|
*,
|
|
engine: Any = engine_client,
|
|
) -> AgenticJobOutcome:
|
|
"""Execute one claimed job without allowing it to block later jobs."""
|
|
|
|
ids = _job_ids(job.job_id)
|
|
try:
|
|
if job.spec.data_classification != DATA_CLASSIFICATION:
|
|
raise continuous_improvement_agentic.AgenticPipelineRejectedError(
|
|
"unsupported scheduled source classification"
|
|
)
|
|
continuous_improvement_agentic.validate_source_packs(job.spec.source_packs)
|
|
if _fingerprint(job.spec) != job.source_fingerprint:
|
|
raise continuous_improvement_agentic.AgenticPipelineRejectedError(
|
|
"scheduled source fingerprint mismatch"
|
|
)
|
|
async with db.acquire(ai_view="research", ai_context=True) as conn:
|
|
result = await continuous_improvement_agentic.run_agentic_content_pipeline(
|
|
conn=conn,
|
|
engine=_ProducerEngine(engine),
|
|
submission_id=ids["submission"],
|
|
pipeline_id=ids["pipeline"],
|
|
benchmark_record_id=ids["benchmark_record"],
|
|
qualification_id=ids["qualification"],
|
|
source_packs=job.spec.source_packs,
|
|
content_kind=job.spec.content_kind,
|
|
difficulty_level=job.spec.difficulty_level,
|
|
variant_count=job.spec.variant_count,
|
|
prompt_version=job.spec.prompt_version,
|
|
trigger_kind=job.spec.trigger_kind,
|
|
)
|
|
if (
|
|
result.state != "pending_human_approval"
|
|
or not result.human_approval_required
|
|
or result.catalog_promoted
|
|
or result.clinical_claim_allowed
|
|
):
|
|
raise continuous_improvement_agentic.AgenticPipelineExecutionError(
|
|
"scheduled pipeline crossed the mandatory human approval boundary"
|
|
)
|
|
await conn.execute(
|
|
"""
|
|
UPDATE app.ci_agentic_job
|
|
SET status = 'completed', lease_started_at = NULL,
|
|
result_submission_id = $2, result_qualification_id = $3,
|
|
completed_at = now(), updated_at = now()
|
|
WHERE job_id = $1 AND status = 'processing'
|
|
""",
|
|
job.job_id,
|
|
result.submission_id,
|
|
result.qualification_id,
|
|
)
|
|
return AgenticJobOutcome(
|
|
job_id=job.job_id,
|
|
status="completed",
|
|
agent_calls_executed=result.agent_calls_executed,
|
|
idempotent_replay=result.idempotent_replay,
|
|
)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except continuous_improvement_agentic.AgenticPipelineRejectedError as exc:
|
|
await _mark_rejected(job.job_id, "safety_gate_rejected", _safe_error_message(exc))
|
|
return AgenticJobOutcome(
|
|
job_id=job.job_id,
|
|
status="rejected",
|
|
agent_calls_executed=0,
|
|
error_code="safety_gate_rejected",
|
|
)
|
|
except continuous_improvement_store.ContinuousImprovementConflictError as exc:
|
|
await _mark_rejected(job.job_id, "idempotency_conflict", _safe_error_message(exc))
|
|
return AgenticJobOutcome(
|
|
job_id=job.job_id,
|
|
status="rejected",
|
|
agent_calls_executed=0,
|
|
error_code="idempotency_conflict",
|
|
)
|
|
except continuous_improvement_agentic.AgenticPipelineExecutionError as exc:
|
|
await _mark_retry(job.job_id, "engine_execution_failed", exc)
|
|
return AgenticJobOutcome(
|
|
job_id=job.job_id,
|
|
status="retry_wait",
|
|
agent_calls_executed=0,
|
|
error_code="engine_execution_failed",
|
|
)
|
|
except Exception as exc:
|
|
await _mark_retry(job.job_id, "unexpected_execution_failed", exc)
|
|
logger.exception("G8 scheduled agentic job failed: job_id=%s", job.job_id)
|
|
return AgenticJobOutcome(
|
|
job_id=job.job_id,
|
|
status="retry_wait",
|
|
agent_calls_executed=0,
|
|
error_code="unexpected_execution_failed",
|
|
)
|
|
|
|
|
|
async def produce_queued_agentic_jobs_once() -> dict[str, int]:
|
|
"""Bootstrap the repo source and process an isolated bounded job batch."""
|
|
|
|
# Imported lazily to keep the trigger's job-spec dependency acyclic. This
|
|
# runs only when the existing default-off producer scheduler (or an explicit
|
|
# one-shot caller) invokes a cycle.
|
|
from . import continuous_improvement_trigger
|
|
|
|
drift_trigger = {
|
|
"drift_signals_scanned": 0,
|
|
"drift_invalid_signals": 0,
|
|
"drift_incidents_created": 0,
|
|
"drift_incident_replays": 0,
|
|
"drift_jobs_enqueued": 0,
|
|
"drift_trigger_failed": 0,
|
|
}
|
|
if settings.continuous_improvement_drift_trigger_enabled:
|
|
try:
|
|
triggered = (
|
|
await continuous_improvement_trigger.enqueue_drift_adversarial_jobs_once()
|
|
)
|
|
drift_trigger.update(
|
|
{
|
|
"drift_signals_scanned": triggered.scanned,
|
|
"drift_invalid_signals": triggered.invalid_signals,
|
|
"drift_incidents_created": triggered.incidents_created,
|
|
"drift_incident_replays": triggered.incident_replays,
|
|
"drift_jobs_enqueued": triggered.jobs_enqueued,
|
|
}
|
|
)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
drift_trigger["drift_trigger_failed"] = 1
|
|
logger.exception("G8 operational drift trigger failed closed")
|
|
|
|
enqueued = 0
|
|
bootstrap_failed = 0
|
|
try:
|
|
await ensure_repo_approved_job()
|
|
enqueued = 1
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
bootstrap_failed = 1
|
|
logger.exception("G8 repo-approved source enqueue failed")
|
|
|
|
counts = {"completed": 0, "retry_wait": 0, "rejected": 0}
|
|
claimed = 0
|
|
for _ in range(settings.continuous_improvement_producer_batch_size):
|
|
try:
|
|
job = await claim_next_agentic_job()
|
|
if job is None:
|
|
break
|
|
claimed += 1
|
|
outcome = await execute_claimed_agentic_job(job)
|
|
counts[outcome.status] += 1
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
counts["retry_wait"] += 1
|
|
logger.exception("G8 scheduler isolated an unhandled job failure")
|
|
return {
|
|
**drift_trigger,
|
|
"repo_source_enqueued": enqueued,
|
|
"bootstrap_failed": bootstrap_failed,
|
|
"claimed": claimed,
|
|
**counts,
|
|
}
|
|
|
|
|
|
async def _producer_loop() -> None:
|
|
delay = settings.continuous_improvement_producer_startup_delay_seconds
|
|
if delay:
|
|
await asyncio.sleep(delay)
|
|
while True:
|
|
try:
|
|
result = await produce_queued_agentic_jobs_once()
|
|
if result["claimed"] or result["bootstrap_failed"]:
|
|
logger.info("G8 scheduled agentic cycle: %s", result)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
logger.exception("G8 scheduled agentic cycle failed")
|
|
await asyncio.sleep(settings.continuous_improvement_producer_interval_seconds)
|
|
|
|
|
|
def schedule_continuous_improvement_producer() -> asyncio.Task[None] | None:
|
|
global _PRODUCER_TASK
|
|
if not settings.continuous_improvement_producer_enabled:
|
|
return None
|
|
if _PRODUCER_TASK is not None and not _PRODUCER_TASK.done():
|
|
return _PRODUCER_TASK
|
|
_PRODUCER_TASK = asyncio.create_task(
|
|
_producer_loop(),
|
|
name="continuous-improvement-agentic-producer",
|
|
)
|
|
return _PRODUCER_TASK
|
|
|
|
|
|
async def stop_continuous_improvement_producer() -> None:
|
|
global _PRODUCER_TASK
|
|
task = _PRODUCER_TASK
|
|
_PRODUCER_TASK = None
|
|
if task is None or task.done():
|
|
return
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
|
|
__all__ = [
|
|
"AgenticJobError",
|
|
"AgenticJobOutcome",
|
|
"ClaimedAgenticJob",
|
|
"ScheduledAgenticJobSpec",
|
|
"claim_next_agentic_job",
|
|
"enqueue_agentic_job",
|
|
"ensure_repo_approved_job",
|
|
"execute_claimed_agentic_job",
|
|
"load_repo_approved_job",
|
|
"produce_queued_agentic_jobs_once",
|
|
"schedule_continuous_improvement_producer",
|
|
"stop_continuous_improvement_producer",
|
|
]
|