e26236fdc1
* Extract session pipeline framework, refactor verification, add UsageProcessor skeleton Pluggable framework under services/session_pipeline/ (contract + lib + per-processor runner) so multiple processors can read /data/user_sessions/<key>/*.jsonl on their own cadence with full failure isolation. Verification flow becomes the first plugin; a no-op UsageProcessor reserves the second slot pending a separate brainstorm on extraction logic + storage shape. Schema v28→v29: rename session_extraction_state → session_processor_state with composite PK (processor_name, session_file). Existing rows copied over with processor_name='verification'; legacy table dropped. Migration is idempotent and no-ops the copy step on fresh installs that came up at the new schema. Endpoint: /api/admin/run-verification-detector replaced by parametrized /api/admin/run-session-processor?processor=<name>. Audit action format follows. Scheduler JOBS: verification-detector entry split into session-processor:verification + session-processor:usage. SCHEDULER_VERIFICATION_DETECTOR_INTERVAL retained for operator compatibility (drives both cadence and health-check grace window); SCHEDULER_USAGE_PROCESSOR_INTERVAL added. * Address PR #232 review: scan dead branch + per-processor lock - `SessionProcessorStateRepository.scan_unprocessed_for` dead else: both branches surfaced every jsonl, the SELECT was unused, runner MD5-rehashed every stable session per tick. Replaced with an mtime precheck — stable sessions (mtime <= processed_at) are filtered at scan; modified files still surface for the runner's authoritative `file_hash` invalidation. Naive-local comparison matches the existing health-check idiom (DuckDB TIMESTAMP strips tz on storage). - Per-processor advisory lock around `_run_processor` in `/api/admin/run-session-processor`. Scheduler tick + manual admin POST could otherwise both run, both call create_evidence on overlapping detections, and accumulate duplicate verification_evidence rows (the dedup short-circuit only covers create+contradiction, not evidence per ADR Decision 3). Non-blocking acquire → 409 Conflict on concurrent invocation; release in finally so a runner exception doesn't wedge the processor. Tests: two new scan unit tests (mtime filter + post-mark mtime bump), 409 endpoint test, lock-released-on-exception test. Two existing tests updated for the new "filtered at scan" stat shape (previously asserted skipped == 1, now scanned == 0). * Address PR #232 review #2: parallel scheduler tick + last_run on terminal state Two pre-existing scaffold bugs in services/scheduler/__main__.py amplified by adding more session-pipeline jobs: 1. Serial for-loop over jobs with synchronous httpx.post(timeout=900) — a 10-minute verification run blocked every other job (data-refresh, health-check, usage, corporate-memory) for the whole window. The PR's stated isolation guarantee held inside the runner but broke at the scheduler dispatch layer. 2. last_run advanced only when _call_api returned True. Permanent-failure jobs hot-looped on every tick (30s) instead of cadence (15min). Fix: ThreadPoolExecutor.submit per due job + per-job in_flight set so a long-running job can't be re-launched on subsequent ticks. last_run advances unconditionally in finally; errors still surface via _call_api logging + audit_log on the receiving side. _run_job extracted to module-level for unit testing. New tests: - TestRunJobBookkeeping: advances on success / failure / unhandled raise - TestRunLoopParallelism: in_flight protection prevents duplicate launches across ticks for a single slow job --------- Co-authored-by: Minas Arustamyan <arustamyan.minas@gmail.com>
55 lines
2 KiB
Python
55 lines
2 KiB
Python
"""Contract for session-pipeline processors.
|
|
|
|
A processor is anything that, given a parsed Claude Code session jsonl file,
|
|
emits some side effect — knowledge extraction, usage events, error metrics,
|
|
security findings, etc. The runner (`services/session_pipeline/runner.py`)
|
|
calls process_session() once per unprocessed file and persists state on success.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Protocol, runtime_checkable
|
|
|
|
import duckdb
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProcessorResult:
|
|
"""Per-session outcome surfaced to the runner. items_count is the number
|
|
of records the processor produced (knowledge items, events, etc.) and
|
|
is stored in session_processor_state.items_extracted for observability —
|
|
not load-bearing for the framework's correctness."""
|
|
items_count: int = 0
|
|
|
|
|
|
@runtime_checkable
|
|
class SessionProcessor(Protocol):
|
|
"""Implementations live in services/session_processors/<name>.py and
|
|
are listed in services/session_processors/__init__.py:PROCESSORS."""
|
|
|
|
name: str
|
|
"""Unique processor key. Used in session_processor_state.processor_name
|
|
and as the URL query param for /api/admin/run-session-processor."""
|
|
|
|
cadence_minutes: int
|
|
"""How often the scheduler should invoke this processor. The actual
|
|
schedule entry is built in services/scheduler/__main__.py from this value
|
|
(env-overridable per processor)."""
|
|
|
|
def process_session(
|
|
self,
|
|
session_path: Path,
|
|
username: str,
|
|
session_key: str,
|
|
conn: duckdb.DuckDBPyConnection,
|
|
) -> ProcessorResult:
|
|
"""Process exactly one session jsonl. Idempotent per
|
|
(name, session_key, file_hash).
|
|
|
|
Raise = the runner will NOT mark this session as processed for this
|
|
processor → it will be retried on the next scheduler tick. Return =
|
|
the runner marks it processed and skips it next time (until its
|
|
file_hash changes)."""
|
|
...
|