Skip to content

Commit 12ed7e9

Browse files
committed
fix(drift): float precision, file-lock atomicity, release wiring
Three issues raised by lan17 in PR review: 1. Float precision on threshold boundary (agentcontrol#1) baseline=1.0, window=0.9, threshold=0.10: IEEE 754 gives drift_magnitude=0.09999999... which fails >= 0.10. Fixed with round(drift_magnitude, 10) >= drift_threshold in _compute_drift(). 2. Race condition on concurrent history writes (agentcontrol#3) load→append→save was not atomic: two workers for the same agent_id would both read stale history and the last writer would silently drop the other's observation. Replaced _load_history() / _save_history() pair with _load_and_append_history() which holds fcntl.LOCK_EX for the full read-modify-write cycle. Lock is per-agent (.lock file), so independent agents remain fully parallel. 3. Release wiring missing for drift package (agentcontrol#2) test-extras, scripts/build.py, Makefile and .PHONY only referenced galileo. Added drift-{test,lint,lint-fix,typecheck,build} targets to Makefile, wired drift-test into test-extras, and added build_evaluator_drift() to scripts/build.py (including 'drift' and 'all' targets).
1 parent ba08a77 commit 12ed7e9

3 files changed

Lines changed: 107 additions & 64 deletions

File tree

‎Makefile‎

Lines changed: 23 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
.PHONY: help sync openapi-spec openapi-spec-check test test-extras test-all models-test test-models test-sdk lint lint-fix typecheck check build build-models build-server build-sdk publish publish-models publish-server publish-sdk hooks-install hooks-uninstall prepush evaluators-test evaluators-lint evaluators-lint-fix evaluators-typecheck evaluators-build galileo-test galileo-lint galileo-lint-fix galileo-typecheck galileo-build sdk-ts-generate sdk-ts-overlay-test sdk-ts-name-check sdk-ts-generate-check sdk-ts-build sdk-ts-test sdk-ts-lint sdk-ts-typecheck sdk-ts-release-check sdk-ts-publish-dry-run sdk-ts-publish
1+
.PHONY: help sync openapi-spec openapi-spec-check test test-extras test-all models-test test-models test-sdk lint lint-fix typecheck check build build-models build-server build-sdk publish publish-models publish-server publish-sdk hooks-install hooks-uninstall prepush evaluators-test evaluators-lint evaluators-lint-fix evaluators-typecheck evaluators-build galileo-test galileo-lint galileo-lint-fix galileo-typecheck galileo-build drift-test drift-lint drift-lint-fix drift-typecheck drift-build sdk-ts-generate sdk-ts-overlay-test sdk-ts-name-check sdk-ts-generate-check sdk-ts-build sdk-ts-test sdk-ts-lint sdk-ts-typecheck sdk-ts-release-check sdk-ts-publish-dry-run sdk-ts-publish
22

33
# Workspace package names
44
PACK_MODELS := agent-control-models
@@ -16,6 +16,7 @@ TS_SDK_DIR := sdks/typescript
1616
ENGINE_DIR := engine
1717
EVALUATORS_DIR := evaluators/builtin
1818
GALILEO_DIR := evaluators/contrib/galileo
19+
DRIFT_DIR := evaluators/contrib/drift
1920
UI_DIR := ui
2021

2122
help:
@@ -33,7 +34,7 @@ help:
3334
@echo "Test:"
3435
@echo " make test - run tests for core packages (models, server, engine, sdk, evaluators)"
3536
@echo " make models-test - run shared model tests with coverage"
36-
@echo " make test-extras - run tests for contrib evaluators (galileo, etc.)"
37+
@echo " make test-extras - run tests for contrib evaluators (galileo, drift, etc.)"
3738
@echo " make test-all - run all tests (core + extras)"
3839
@echo " make sdk-ts-test - run TypeScript SDK tests"
3940
@echo ""
@@ -90,7 +91,7 @@ models-test:
9091
test-models: models-test
9192

9293
# Run tests for contrib evaluators (not included in default test target)
93-
test-extras: galileo-test
94+
test-extras: galileo-test drift-test
9495

9596
# Run all tests (core + extras)
9697
test-all: test test-extras
@@ -244,3 +245,22 @@ galileo-typecheck:
244245

245246
galileo-build:
246247
$(MAKE) -C $(GALILEO_DIR) build
248+
249+
# ---------------------------
250+
# Contrib Evaluators (Drift)
251+
# ---------------------------
252+
253+
drift-test:
254+
$(MAKE) -C $(DRIFT_DIR) test
255+
256+
drift-lint:
257+
$(MAKE) -C $(DRIFT_DIR) lint
258+
259+
drift-lint-fix:
260+
$(MAKE) -C $(DRIFT_DIR) lint-fix
261+
262+
drift-typecheck:
263+
$(MAKE) -C $(DRIFT_DIR) typecheck
264+
265+
drift-build:
266+
$(MAKE) -C $(DRIFT_DIR) build

‎evaluators/contrib/drift/src/agent_control_evaluator_drift/drift/evaluator.py‎

Lines changed: 60 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66

77
from __future__ import annotations
88

9+
import fcntl
910
import json
1011
import logging
1112
import os
@@ -20,40 +21,58 @@
2021
logger = logging.getLogger(__name__)
2122

2223

23-
def _load_history(path: Path) -> list[float]:
24-
"""Load observation history from a JSON file.
24+
def _load_and_append_history(path: Path, score: float) -> list[float]:
25+
"""Atomically load history, append a score, persist, and return the updated list.
26+
27+
Uses an exclusive advisory lock (``fcntl.LOCK_EX``) on the history file so
28+
that concurrent workers for the same agent do not race on the
29+
read-modify-write cycle. Without this, two simultaneous calls can both read
30+
the same stale list and the last writer silently drops the other's
31+
observation, causing drift detection to miss events.
2532
2633
Args:
27-
path: Path to the history file.
34+
path: Path to the per-agent JSON history file.
35+
score: New observation to append (already validated, in [0.0, 1.0]).
2836
2937
Returns:
30-
List of float scores, or empty list if file doesn't exist.
31-
"""
32-
if not path.exists():
33-
return []
34-
try:
35-
with path.open("r") as fh:
36-
data = json.load(fh)
37-
scores = data.get("scores", [])
38-
return [float(s) for s in scores if isinstance(s, (int, float))]
39-
except (json.JSONDecodeError, OSError, ValueError) as exc:
40-
logger.warning("Failed to load drift history from %s: %s", path, exc)
41-
return []
42-
43-
44-
def _save_history(path: Path, scores: list[float]) -> None:
45-
"""Persist observation history to a JSON file.
46-
47-
Args:
48-
path: Path to the history file.
49-
scores: List of float scores to persist.
38+
Updated list of float scores (oldest first), including *score*.
5039
5140
Raises:
52-
OSError: If the file cannot be written.
41+
OSError: If the lock file or history file cannot be opened or written.
42+
json.JSONDecodeError: If the history file contains malformed JSON.
5343
"""
5444
path.parent.mkdir(parents=True, exist_ok=True)
55-
with path.open("w") as fh:
56-
json.dump({"scores": scores}, fh)
45+
lock_path = path.with_suffix(".lock")
46+
47+
# Open (or create) the lock file and hold an exclusive lock for the
48+
# duration of the read-modify-write. The lock is released automatically
49+
# when the file descriptor is closed at the end of this block.
50+
with lock_path.open("a") as lock_fh:
51+
fcntl.flock(lock_fh, fcntl.LOCK_EX)
52+
53+
# Read existing scores under the lock.
54+
scores: list[float] = []
55+
if path.exists():
56+
try:
57+
with path.open("r") as fh:
58+
data = json.load(fh)
59+
scores = [
60+
float(s)
61+
for s in data.get("scores", [])
62+
if isinstance(s, (int, float))
63+
]
64+
except (json.JSONDecodeError, ValueError) as exc:
65+
logger.warning("Corrupt drift history at %s; resetting. Error: %s", path, exc)
66+
scores = []
67+
68+
# Append and persist while the lock is still held.
69+
scores.append(score)
70+
with path.open("w") as fh:
71+
json.dump({"scores": scores}, fh)
72+
73+
# Lock released here when lock_fh closes.
74+
75+
return scores
5776

5877

5978
def _compute_drift(
@@ -109,7 +128,11 @@ def _compute_drift(
109128
recent_avg = sum(recent_scores) / len(recent_scores)
110129
drift_magnitude = baseline_avg - recent_avg # positive = drop
111130

112-
matched = drift_magnitude >= drift_threshold
131+
# Round before threshold comparison to avoid float precision issues.
132+
# e.g. 1.0 - 0.9 evaluates to 0.09999999... in IEEE 754, which fails a
133+
# >= 0.10 check without rounding. Rounding to 10 decimal places preserves
134+
# all meaningful precision while eliminating the ULP-level noise.
135+
matched = round(drift_magnitude, 10) >= drift_threshold
113136

114137
if matched:
115138
status = "drift_detected"
@@ -152,12 +175,12 @@ class DriftEvaluator(Evaluator[DriftEvaluatorConfig]):
152175
153176
No external API or service required — history is stored as local JSON.
154177
155-
Instance Caching Note:
156-
Per the base class contract, this evaluator stores only immutable
157-
config state in ``__init__``. All file I/O happens inside
158-
``evaluate()`` using local variables, making it safe to reuse
159-
across concurrent requests (each call reads and writes atomically
160-
via a per-agent file lock-free JSON write).
178+
Concurrency:
179+
Each ``evaluate()`` call uses an exclusive advisory file lock
180+
(``fcntl.LOCK_EX``) scoped to the read-modify-write cycle, ensuring
181+
that concurrent workers for the same ``agent_id`` never race on
182+
history updates. The lock is per-agent (``<agent_id>.lock`` next to
183+
``<agent_id>.json``), so different agents remain fully parallel.
161184
162185
Example:
163186
```python
@@ -222,45 +245,23 @@ async def evaluate(self, data: Any) -> EvaluatorResult:
222245
storage_dir = Path(self.config.storage_path)
223246
history_path = storage_dir / f"{self.config.agent_id}.json"
224247

225-
# Load existing history
248+
# Atomically load, append, and persist history under a file lock.
226249
try:
227-
scores = _load_history(history_path)
250+
scores = _load_and_append_history(history_path, score)
228251
except Exception as exc:
229-
logger.error("DriftEvaluator: failed to load history: %s", exc)
252+
logger.error("DriftEvaluator: storage error: %s", exc)
230253
matched = self.config.on_error == "deny"
231254
return EvaluatorResult(
232255
matched=matched,
233256
confidence=0.0,
234-
message=f"Storage error (load): {exc}",
257+
message=f"Storage error: {exc}",
235258
metadata={
236259
"error": str(exc),
237260
"agent_id": self.config.agent_id,
238261
"fallback_action": self.config.on_error,
239262
},
240263
)
241264

242-
# Append new score
243-
scores.append(score)
244-
245-
# Persist updated history
246-
try:
247-
_save_history(history_path, scores)
248-
except Exception as exc:
249-
logger.error("DriftEvaluator: failed to save history: %s", exc)
250-
# Still compute drift from in-memory scores even if save fails
251-
matched_on_error = self.config.on_error == "deny"
252-
if matched_on_error:
253-
return EvaluatorResult(
254-
matched=True,
255-
confidence=0.0,
256-
message=f"Storage error (save): {exc}",
257-
metadata={
258-
"error": str(exc),
259-
"agent_id": self.config.agent_id,
260-
"fallback_action": self.config.on_error,
261-
},
262-
)
263-
264265
# Compute drift metrics
265266
metrics = _compute_drift(
266267
scores=scores,

‎scripts/build.py‎

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
afterward. This allows the published wheels to be self-contained.
77
88
Usage:
9-
python scripts/build.py [models|evaluators|sdk|server|galileo|all]
9+
python scripts/build.py [models|evaluators|sdk|server|galileo|drift|all]
1010
"""
1111

1212
import shutil
@@ -222,6 +222,25 @@ def build_evaluator_galileo() -> None:
222222
print(f" Built agent-control-evaluator-galileo v{version}")
223223

224224

225+
def build_evaluator_drift() -> None:
226+
"""Build agent-control-evaluator-drift (standalone, no vendoring needed)."""
227+
version = get_global_version()
228+
drift_dir = ROOT / "evaluators" / "contrib" / "drift"
229+
230+
print(f"Building agent-control-evaluator-drift v{version}")
231+
232+
# Clean previous builds
233+
dist_dir = drift_dir / "dist"
234+
if dist_dir.exists():
235+
shutil.rmtree(dist_dir)
236+
237+
# Set version
238+
set_package_version(drift_dir / "pyproject.toml", version)
239+
240+
subprocess.run(["uv", "build", "-o", str(dist_dir)], cwd=drift_dir, check=True)
241+
print(f" Built agent-control-evaluator-drift v{version}")
242+
243+
225244
def build_all() -> None:
226245
"""Build all packages."""
227246
print(f"Building all packages (version {get_global_version()})\n")
@@ -230,6 +249,7 @@ def build_all() -> None:
230249
build_sdk()
231250
build_server()
232251
build_evaluator_galileo()
252+
build_evaluator_drift()
233253
print("\nAll packages built successfully!")
234254

235255

@@ -248,8 +268,10 @@ def build_all() -> None:
248268
build_server()
249269
elif target == "galileo":
250270
build_evaluator_galileo()
271+
elif target == "drift":
272+
build_evaluator_drift()
251273
elif target == "all":
252274
build_all()
253275
else:
254-
print("Usage: python scripts/build.py [models|evaluators|sdk|server|galileo|all]")
276+
print("Usage: python scripts/build.py [models|evaluators|sdk|server|galileo|drift|all]")
255277
sys.exit(1)

0 commit comments

Comments
 (0)