Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/jean/agent/capture_clipboard.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
19 changes: 15 additions & 4 deletions src/jean/agent/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import asyncio
import logging
import os
import threading
import uuid

import structlog
Expand Down Expand Up @@ -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)
Expand All @@ -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)

Expand Down
7 changes: 6 additions & 1 deletion src/jean/aggregator/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion src/jean/bridges/flowfabric.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion src/jean/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
)
Expand Down
26 changes: 23 additions & 3 deletions src/jean/process_mapper/drift_detector.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
10 changes: 10 additions & 0 deletions src/jean/validator/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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={
Expand Down Expand Up @@ -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={
Expand Down
7 changes: 4 additions & 3 deletions tests/test_drift_detector.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 == []

Expand Down
Loading