From 390945e5d2ffac9939085b4647ac66450bf4111a Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 25 Mar 2026 16:54:34 +0000 Subject: [PATCH] fix: resolve 7 bugs found in codebase analysis MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CRITICAL: - main.py: replace bare list[str] current_app with threading.Lock-protected accessors to eliminate race condition between asyncio loop and pynput thread - aggregator/api.py: add add_done_callback error handler on fire-and-forget observation dispatch tasks so unhandled exceptions are logged - validator/api.py: enforce OBSERVED→VALIDATED state machine — approve endpoint (REST + FlowFabric webhook) now returns 409 if observation is not OBSERVED - drift_detector.py: return drift_score=0.0 when traces list is empty; no observed baseline means drift cannot be measured (was incorrectly 1.0) MAJOR: - models.py: ProcessStep.related_event_types changed from list[str] to list[EventType] for compile-time type safety; Pydantic v2 coerces strings - capture_clipboard.py: paste payload from_app now "" instead of None when no prior copy was detected, preventing None in serialized payloads - flowfabric.py: validated_by defaults to "" in payload instead of None tests: update test_empty_traces to reflect corrected zero-drift semantics --- src/jean/agent/capture_clipboard.py | 2 +- src/jean/agent/main.py | 19 +++++++++++++---- src/jean/aggregator/api.py | 7 +++++- src/jean/bridges/flowfabric.py | 2 +- src/jean/models.py | 2 +- src/jean/process_mapper/drift_detector.py | 26 ++++++++++++++++++++--- src/jean/validator/api.py | 10 +++++++++ tests/test_drift_detector.py | 7 +++--- 8 files changed, 61 insertions(+), 14 deletions(-) diff --git a/src/jean/agent/capture_clipboard.py b/src/jean/agent/capture_clipboard.py index a836b40..840767a 100644 --- a/src/jean/agent/capture_clipboard.py +++ b/src/jean/agent/capture_clipboard.py @@ -148,7 +148,7 @@ def _check_clipboard_shortcuts(self) -> None: paste_event = self._make_event( EventType.CLIPBOARD_PASTE, current_app, - {"from_app": self._last_copy_app, "is_cross_app": is_cross_app}, + {"from_app": self._last_copy_app or "", "is_cross_app": is_cross_app}, ) self._emit(paste_event) diff --git a/src/jean/agent/main.py b/src/jean/agent/main.py index 39d23b3..b0bd8ef 100644 --- a/src/jean/agent/main.py +++ b/src/jean/agent/main.py @@ -17,6 +17,7 @@ import asyncio import logging import os +import threading import uuid import structlog @@ -50,8 +51,18 @@ async def _capture_loop( kb_queue: asyncio.Queue = asyncio.Queue() loop = asyncio.get_running_loop() - # Shared mutable container for current active app (updated by AppTransitionCapture drain) - current_app: list[str] = ["unknown"] + # Thread-safe container for current active app. + # Written by the asyncio drain coroutine, read by the pynput background thread. + _current_app_lock = threading.Lock() + _current_app_value: list[str] = ["unknown"] + + def _get_current_app() -> str: + with _current_app_lock: + return _current_app_value[0] + + def _set_current_app(app: str) -> None: + with _current_app_lock: + _current_app_value[0] = app app_cap = AppTransitionCapture(workstation_id, process_context, session_id) struct_cap = StructuralActionCapture(workstation_id, process_context, session_id) @@ -67,14 +78,14 @@ async def _capture_loop( workstation_id=workstation_id, process_context=process_context, session_id=session_id, - current_app_fn=lambda: current_app[0], + current_app_fn=_get_current_app, loop=loop, ) async def _drain_gen(cap: AppTransitionCapture | StructuralActionCapture) -> None: async for event in cap.events(): if event.app != "unknown": - current_app[0] = event.app + _set_current_app(event.app) await buffer.push(event) log.debug("Captured event", type=event.type, app=event.app) diff --git a/src/jean/aggregator/api.py b/src/jean/aggregator/api.py index cc68565..be836e3 100644 --- a/src/jean/aggregator/api.py +++ b/src/jean/aggregator/api.py @@ -100,7 +100,12 @@ async def ingest(events: list[BusinessEvent], _: None = Depends(require_auth)) - # Auto-generate FieldObservations and dispatch in the background (fire-and-forget) observations = _obs_generator.generate(_patterns) for obs in observations: - asyncio.create_task(_obs_dispatcher.dispatch(obs)) + task = asyncio.create_task(_obs_dispatcher.dispatch(obs)) + task.add_done_callback( + lambda t: log.warning("Observation dispatch failed", error=str(t.exception())) + if not t.cancelled() and t.exception() is not None + else None + ) # Run irritant detection on the newly ingested traces irritant_signals = _irritant_detector.detect(new_traces) diff --git a/src/jean/bridges/flowfabric.py b/src/jean/bridges/flowfabric.py index 96e0fba..45688b8 100644 --- a/src/jean/bridges/flowfabric.py +++ b/src/jean/bridges/flowfabric.py @@ -81,7 +81,7 @@ def _build_payload(self, observation: FieldObservation) -> dict: "observed_behavior": observation.observed_behavior, "gap_score": observation.gap_score, "supporting_patterns": observation.supporting_patterns, - "validated_by": observation.validated_by, + "validated_by": observation.validated_by or "", "validated_at": ( observation.validated_at.isoformat() if observation.validated_at diff --git a/src/jean/models.py b/src/jean/models.py index a6fb967..0da8a4b 100644 --- a/src/jean/models.py +++ b/src/jean/models.py @@ -287,7 +287,7 @@ class ProcessStep(BaseModel): le=1.0, description="Friction score: 0=smooth, 1=highly frustrating", ) - related_event_types: list[str] = Field( + related_event_types: list[EventType] = Field( default_factory=list, description="EventType values observed at this step", ) diff --git a/src/jean/process_mapper/drift_detector.py b/src/jean/process_mapper/drift_detector.py index 6a98b5d..7657633 100644 --- a/src/jean/process_mapper/drift_detector.py +++ b/src/jean/process_mapper/drift_detector.py @@ -39,11 +39,31 @@ def compute( process: ProcessDefinition, traces: list[SessionTrace], ) -> DriftReport: - """Compute a DriftReport for the given process and its matching traces.""" - # Declared event types from steps + """Compute a DriftReport for the given process and its matching traces. + + Returns drift_score=0.0 when no traces are available — there is no + observed baseline to compare against, so drift cannot be measured. + """ + if not traces: + return DriftReport( + process_id=process.id, + process_name=process.name, + process_context=process.process_context, + declared_step_count=len(process.steps), + observed_event_types=[], + declared_event_types=sorted( + {et.value for step in process.steps for et in step.related_event_types} + ), + drift_score=0.0, + session_count_analysed=0, + computed_at=datetime.now(timezone.utc), + alert=False, + ) + + # Declared event types from steps (use .value for uniform string comparison) declared: set[str] = set() for step in process.steps: - declared.update(step.related_event_types) + declared.update(et.value for et in step.related_event_types) # Observed event types from traces observed: set[str] = set() diff --git a/src/jean/validator/api.py b/src/jean/validator/api.py index a071078..d695d80 100644 --- a/src/jean/validator/api.py +++ b/src/jean/validator/api.py @@ -163,6 +163,11 @@ async def approve_observation( if obs.state == ProcedureState.VALIDATED: raise HTTPException(status_code=409, detail="Observation is already VALIDATED") + if obs.state != ProcedureState.OBSERVED: + raise HTTPException( + status_code=409, + detail=f"Only OBSERVED observations can be approved (current state: {obs.state})", + ) validated = obs.model_copy( update={ @@ -251,6 +256,11 @@ async def flowfabric_webhook(req: FlowFabricWebhookRequest) -> FieldObservation: if req.action == _WebhookAction.APPROVE: if obs.state == ProcedureState.VALIDATED: raise HTTPException(status_code=409, detail="Observation is already VALIDATED") + if obs.state != ProcedureState.OBSERVED: + raise HTTPException( + status_code=409, + detail=f"Only OBSERVED observations can be approved (current state: {obs.state})", + ) validated = obs.model_copy( update={ diff --git a/tests/test_drift_detector.py b/tests/test_drift_detector.py index eb89a8c..6e20bb5 100644 --- a/tests/test_drift_detector.py +++ b/tests/test_drift_detector.py @@ -128,12 +128,13 @@ def test_no_alert_below_threshold(): def test_empty_traces(): - """No traces → observed set is empty; drift based on declared types only.""" + """No traces → no observed baseline; drift_score=0.0 (cannot be measured).""" process = _make_process(related_event_types=[[EventType.SAVE]]) detector = DriftDetector() report = detector.compute(process, traces=[]) - # declared={save}, observed={}, union={save}, intersection={} → score=1.0 - assert report.drift_score == 1.0 + # No data to compare against → score=0.0, alert=False + assert report.drift_score == 0.0 + assert report.alert is False assert report.session_count_analysed == 0 assert report.observed_event_types == []