gabriel / muse public
test_harmony_phase4.py python
694 lines 25.7 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 142 days ago
1 """Tests for Phase 4 — Escalation Records.
2
3 New additions to ``muse/core/harmony.py``:
4
5 ``EscalationStatus`` — open string constants: OPEN, RESOLVED
6 ``AuditEventType.ESCALATION_RESOLVED``
7 ``EscalationRecord`` — frozen dataclass
8 ``compute_escalation_id`` — deterministic hex64 from (pattern_id, reason)
9 ``escalations_dir`` — ``.muse/harmony/escalations/``
10 ``record_escalation`` — atomic write; idempotent
11 ``load_escalation`` — read by ID
12 ``list_escalations`` — filtered list (status filter)
13 ``resolve_escalation`` — transition open → resolved, atomic
14
15 Coverage tiers
16 --------------
17 I Unit — EscalationStatus constants, EscalationRecord fields,
18 compute_escalation_id determinism + collision resistance
19 II Integration — record / load / list / resolve_escalation CRUD
20 III End-to-end — escalate → list → resolve → audit trail
21 IV Stress — 200-escalation list, concurrent record (same ID idempotency)
22 V Data integrity — JSON round-trip, all fields always present, frozen
23 VI Security — ID validation, path traversal rejected, size cap respected
24 VII Performance — all ops <50 ms
25 """
26 from __future__ import annotations
27
28 import concurrent.futures
29 import dataclasses
30 import datetime
31 import hashlib
32 import json
33 import pathlib
34 import time
35
36 import pytest
37
38 import muse.core.harmony as h
39 from muse.core.harmony import (
40 AgentProvenance,
41 AuditEventType,
42 EscalationRecord,
43 EscalationStatus,
44 append_audit,
45 blob_fingerprint,
46 compute_escalation_id,
47 compute_pattern_id,
48 compute_resolution_id,
49 escalations_dir,
50 list_audit,
51 list_escalations,
52 load_escalation,
53 record_escalation,
54 resolve_escalation,
55 save_resolution,
56 Resolution,
57 )
58
59
60 # ---------------------------------------------------------------------------
61 # Shared helpers
62 # ---------------------------------------------------------------------------
63
64
65 def _fake_id(seed: str) -> str:
66 return hashlib.sha256(seed.encode()).hexdigest()
67
68
69 def _utc_now() -> datetime.datetime:
70 return datetime.datetime.now(datetime.timezone.utc)
71
72
73 @pytest.fixture()
74 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
75 (tmp_path / ".muse").mkdir()
76 return tmp_path
77
78
79 def _make_record(
80 pattern_id: str | None = None,
81 reason: str = "No match found",
82 agent_id: str | None = None,
83 ) -> EscalationRecord:
84 pid = pattern_id or _fake_id("pat1")
85 eid = compute_escalation_id(pid, reason)
86 return EscalationRecord(
87 escalation_id=eid,
88 pattern_id=pid,
89 reason=reason,
90 escalated_at=_utc_now(),
91 escalated_by=(
92 AgentProvenance.agent(agent_id) if agent_id else AgentProvenance.human()
93 ),
94 )
95
96
97 # ===========================================================================
98 # Tier I — Unit
99 # ===========================================================================
100
101
102 class TestEscalationStatusConstants:
103 """I: EscalationStatus is an open string-constant namespace."""
104
105 def test_open_value(self) -> None:
106 assert EscalationStatus.OPEN == "open"
107
108 def test_resolved_value(self) -> None:
109 assert EscalationStatus.RESOLVED == "resolved"
110
111 def test_constants_are_strings(self) -> None:
112 assert isinstance(EscalationStatus.OPEN, str)
113 assert isinstance(EscalationStatus.RESOLVED, str)
114
115
116 class TestAuditEventTypeEscalationResolved:
117 """I: AuditEventType.ESCALATION_RESOLVED constant exists."""
118
119 def test_escalation_resolved_constant(self) -> None:
120 assert AuditEventType.ESCALATION_RESOLVED == "escalation_resolved"
121
122 def test_escalation_recorded_still_present(self) -> None:
123 assert AuditEventType.ESCALATION_RECORDED == "escalation_recorded"
124
125
126 class TestEscalationRecordFields:
127 """I: EscalationRecord dataclass shape and defaults."""
128
129 def test_required_fields_present(self) -> None:
130 pid = _fake_id("p")
131 eid = compute_escalation_id(pid, "reason")
132 r = EscalationRecord(
133 escalation_id=eid,
134 pattern_id=pid,
135 reason="test reason",
136 escalated_at=_utc_now(),
137 escalated_by=AgentProvenance.human(),
138 )
139 assert r.escalation_id == eid
140 assert r.pattern_id == pid
141 assert r.reason == "test reason"
142 assert r.status == EscalationStatus.OPEN
143
144 def test_status_defaults_to_open(self) -> None:
145 r = _make_record()
146 assert r.status == EscalationStatus.OPEN
147
148 def test_optional_fields_default_none(self) -> None:
149 r = _make_record()
150 assert r.resolved_at is None
151 assert r.resolved_by is None
152 assert r.resolution_id is None
153
154 def test_frozen(self) -> None:
155 r = _make_record()
156 with pytest.raises((dataclasses.FrozenInstanceError, AttributeError)):
157 r.status = EscalationStatus.RESOLVED # type: ignore[misc]
158
159 def test_is_dataclass(self) -> None:
160 assert dataclasses.is_dataclass(EscalationRecord)
161
162
163 class TestComputeEscalationId:
164 """I: compute_escalation_id is deterministic and collision-resistant."""
165
166 def test_deterministic_same_inputs(self) -> None:
167 pid = _fake_id("p")
168 eid1 = compute_escalation_id(pid, "reason")
169 eid2 = compute_escalation_id(pid, "reason")
170 assert eid1 == eid2
171
172 def test_different_pattern_ids_differ(self) -> None:
173 pid1 = _fake_id("p1")
174 pid2 = _fake_id("p2")
175 assert compute_escalation_id(pid1, "reason") != compute_escalation_id(pid2, "reason")
176
177 def test_different_reasons_differ(self) -> None:
178 pid = _fake_id("p")
179 assert compute_escalation_id(pid, "reason A") != compute_escalation_id(pid, "reason B")
180
181 def test_returns_hex64(self) -> None:
182 eid = compute_escalation_id(_fake_id("p"), "reason")
183 assert len(eid) == 64
184 assert all(c in "0123456789abcdef" for c in eid)
185
186
187 # ===========================================================================
188 # Tier II — Integration: CRUD
189 # ===========================================================================
190
191
192 class TestEscalationsDir:
193 """II: escalations_dir returns the correct path."""
194
195 def test_path_structure(self, repo: pathlib.Path) -> None:
196 d = escalations_dir(repo)
197 assert d == repo / ".muse" / "harmony" / "escalations"
198
199 def test_does_not_create_dir(self, repo: pathlib.Path) -> None:
200 d = escalations_dir(repo)
201 assert not d.exists()
202
203
204 class TestRecordEscalation:
205 """II: record_escalation writes a JSON file; idempotent."""
206
207 def test_creates_escalation_file(self, repo: pathlib.Path) -> None:
208 rec = _make_record()
209 record_escalation(repo, rec)
210 dest = escalations_dir(repo) / f"{rec.escalation_id}.json"
211 assert dest.exists()
212
213 def test_idempotent_second_write_no_error(self, repo: pathlib.Path) -> None:
214 rec = _make_record()
215 record_escalation(repo, rec)
216 record_escalation(repo, rec) # should not raise
217 dest = escalations_dir(repo) / f"{rec.escalation_id}.json"
218 assert dest.exists()
219
220 def test_idempotent_returns_false_on_second_write(self, repo: pathlib.Path) -> None:
221 rec = _make_record()
222 first = record_escalation(repo, rec)
223 second = record_escalation(repo, rec)
224 assert first is True
225 assert second is False
226
227 def test_json_contains_expected_keys(self, repo: pathlib.Path) -> None:
228 rec = _make_record()
229 record_escalation(repo, rec)
230 dest = escalations_dir(repo) / f"{rec.escalation_id}.json"
231 data = json.loads(dest.read_text())
232 for key in ("escalation_id", "pattern_id", "reason", "escalated_at",
233 "escalated_by", "status"):
234 assert key in data
235
236 def test_status_stored_as_open(self, repo: pathlib.Path) -> None:
237 rec = _make_record()
238 record_escalation(repo, rec)
239 dest = escalations_dir(repo) / f"{rec.escalation_id}.json"
240 data = json.loads(dest.read_text())
241 assert data["status"] == EscalationStatus.OPEN
242
243 def test_atomic_write_no_temp_files(self, repo: pathlib.Path) -> None:
244 rec = _make_record()
245 record_escalation(repo, rec)
246 esc_dir = escalations_dir(repo)
247 tmp_files = list(esc_dir.glob("*.tmp"))
248 assert tmp_files == []
249
250
251 class TestLoadEscalation:
252 """II: load_escalation retrieves a stored record."""
253
254 def test_returns_none_for_missing(self, repo: pathlib.Path) -> None:
255 eid = _fake_id("missing")
256 assert load_escalation(repo, eid) is None
257
258 def test_round_trips_required_fields(self, repo: pathlib.Path) -> None:
259 rec = _make_record()
260 record_escalation(repo, rec)
261 loaded = load_escalation(repo, rec.escalation_id)
262 assert loaded is not None
263 assert loaded.escalation_id == rec.escalation_id
264 assert loaded.pattern_id == rec.pattern_id
265 assert loaded.reason == rec.reason
266 assert loaded.status == EscalationStatus.OPEN
267
268 def test_escalated_by_human_round_trips(self, repo: pathlib.Path) -> None:
269 rec = _make_record()
270 record_escalation(repo, rec)
271 loaded = load_escalation(repo, rec.escalation_id)
272 assert loaded is not None
273 assert loaded.escalated_by.type == "human"
274
275 def test_escalated_by_agent_round_trips(self, repo: pathlib.Path) -> None:
276 rec = _make_record(agent_id="claude-code")
277 record_escalation(repo, rec)
278 loaded = load_escalation(repo, rec.escalation_id)
279 assert loaded is not None
280 assert loaded.escalated_by.agent_id == "claude-code"
281
282 def test_validates_id_rejects_traversal(self, repo: pathlib.Path) -> None:
283 with pytest.raises(ValueError):
284 load_escalation(repo, "../../evil")
285
286
287 class TestListEscalations:
288 """II: list_escalations returns sorted list with optional status filter."""
289
290 def test_empty_store_returns_empty(self, repo: pathlib.Path) -> None:
291 assert list_escalations(repo) == []
292
293 def test_returns_all_by_default(self, repo: pathlib.Path) -> None:
294 pid1, pid2 = _fake_id("p1"), _fake_id("p2")
295 record_escalation(repo, _make_record(pid1, "r1"))
296 record_escalation(repo, _make_record(pid2, "r2"))
297 assert len(list_escalations(repo)) == 2
298
299 def test_filter_open_only(self, repo: pathlib.Path) -> None:
300 pid1, pid2 = _fake_id("p1"), _fake_id("p2")
301 rec1 = _make_record(pid1, "r1")
302 rec2 = _make_record(pid2, "r2")
303 record_escalation(repo, rec1)
304 record_escalation(repo, rec2)
305 resolve_escalation(
306 repo, rec2.escalation_id, _fake_id("res"),
307 AgentProvenance.human(), _utc_now()
308 )
309 results = list_escalations(repo, status=EscalationStatus.OPEN)
310 assert len(results) == 1
311 assert results[0].escalation_id == rec1.escalation_id
312
313 def test_filter_resolved_only(self, repo: pathlib.Path) -> None:
314 pid1, pid2 = _fake_id("p1"), _fake_id("p2")
315 rec1 = _make_record(pid1, "r1")
316 rec2 = _make_record(pid2, "r2")
317 record_escalation(repo, rec1)
318 record_escalation(repo, rec2)
319 resolve_escalation(
320 repo, rec2.escalation_id, _fake_id("res"),
321 AgentProvenance.human(), _utc_now()
322 )
323 results = list_escalations(repo, status=EscalationStatus.RESOLVED)
324 assert len(results) == 1
325 assert results[0].escalation_id == rec2.escalation_id
326
327 def test_sorted_newest_first(self, repo: pathlib.Path) -> None:
328 pids = [_fake_id(f"p{i}") for i in range(3)]
329 for i, pid in enumerate(pids):
330 rec = _make_record(pid, f"reason {i}")
331 record_escalation(repo, rec)
332 results = list_escalations(repo)
333 timestamps = [r.escalated_at for r in results]
334 assert timestamps == sorted(timestamps, reverse=True)
335
336 def test_skips_symlinks(self, repo: pathlib.Path) -> None:
337 rec = _make_record()
338 record_escalation(repo, rec)
339 esc_dir = escalations_dir(repo)
340 link = esc_dir / ("a" * 64 + ".json")
341 link.symlink_to(esc_dir / f"{rec.escalation_id}.json")
342 results = list_escalations(repo)
343 assert len(results) == 1 # symlink skipped
344
345
346 class TestResolveEscalation:
347 """II: resolve_escalation transitions open → resolved atomically."""
348
349 def test_returns_true_when_found(self, repo: pathlib.Path) -> None:
350 rec = _make_record()
351 record_escalation(repo, rec)
352 result = resolve_escalation(
353 repo, rec.escalation_id, _fake_id("res"),
354 AgentProvenance.human(), _utc_now()
355 )
356 assert result is True
357
358 def test_returns_false_when_not_found(self, repo: pathlib.Path) -> None:
359 eid = _fake_id("missing")
360 result = resolve_escalation(
361 repo, eid, _fake_id("res"),
362 AgentProvenance.human(), _utc_now()
363 )
364 assert result is False
365
366 def test_status_updated_to_resolved(self, repo: pathlib.Path) -> None:
367 rec = _make_record()
368 record_escalation(repo, rec)
369 resolve_escalation(
370 repo, rec.escalation_id, _fake_id("res"),
371 AgentProvenance.human(), _utc_now()
372 )
373 loaded = load_escalation(repo, rec.escalation_id)
374 assert loaded is not None
375 assert loaded.status == EscalationStatus.RESOLVED
376
377 def test_resolution_id_stored(self, repo: pathlib.Path) -> None:
378 rec = _make_record()
379 record_escalation(repo, rec)
380 res_id = _fake_id("resolution")
381 resolve_escalation(
382 repo, rec.escalation_id, res_id,
383 AgentProvenance.human(), _utc_now()
384 )
385 loaded = load_escalation(repo, rec.escalation_id)
386 assert loaded is not None
387 assert loaded.resolution_id == res_id
388
389 def test_resolved_by_stored(self, repo: pathlib.Path) -> None:
390 rec = _make_record()
391 record_escalation(repo, rec)
392 actor = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
393 resolve_escalation(repo, rec.escalation_id, _fake_id("res"), actor, _utc_now())
394 loaded = load_escalation(repo, rec.escalation_id)
395 assert loaded is not None
396 assert loaded.resolved_by is not None
397 assert loaded.resolved_by.agent_id == "claude-code"
398
399 def test_resolved_at_stored(self, repo: pathlib.Path) -> None:
400 rec = _make_record()
401 record_escalation(repo, rec)
402 now = _utc_now()
403 resolve_escalation(repo, rec.escalation_id, _fake_id("res"),
404 AgentProvenance.human(), now)
405 loaded = load_escalation(repo, rec.escalation_id)
406 assert loaded is not None
407 assert loaded.resolved_at is not None
408
409 def test_validates_escalation_id(self, repo: pathlib.Path) -> None:
410 with pytest.raises(ValueError):
411 resolve_escalation(
412 repo, "bad-id", _fake_id("res"),
413 AgentProvenance.human(), _utc_now()
414 )
415
416 def test_no_temp_files_after_resolve(self, repo: pathlib.Path) -> None:
417 rec = _make_record()
418 record_escalation(repo, rec)
419 resolve_escalation(
420 repo, rec.escalation_id, _fake_id("res"),
421 AgentProvenance.human(), _utc_now()
422 )
423 esc_dir = escalations_dir(repo)
424 assert list(esc_dir.glob("*.tmp")) == []
425
426
427 # ===========================================================================
428 # Tier III — End-to-end
429 # ===========================================================================
430
431
432 class TestEndToEnd:
433 """III: full lifecycle: escalate → list → resolve → audit."""
434
435 def test_escalate_list_resolve_audit(self, repo: pathlib.Path) -> None:
436 pid = _fake_id("track.mid:content")
437 reason = "No policy or resolution found for pattern"
438 actor = AgentProvenance.agent("claude-code")
439
440 # 1. record escalation
441 rec = EscalationRecord(
442 escalation_id=compute_escalation_id(pid, reason),
443 pattern_id=pid,
444 reason=reason,
445 escalated_at=_utc_now(),
446 escalated_by=actor,
447 )
448 record_escalation(repo, rec)
449
450 # 2. audit escalation event
451 append_audit(repo, AuditEventType.ESCALATION_RECORDED, actor, pattern_id=pid)
452
453 # 3. list shows 1 open escalation
454 open_esc = list_escalations(repo, status=EscalationStatus.OPEN)
455 assert len(open_esc) == 1
456
457 # 4. resolve it
458 res_id = _fake_id("resolution")
459 resolve_escalation(repo, rec.escalation_id, res_id, actor, _utc_now())
460 append_audit(
461 repo, AuditEventType.ESCALATION_RESOLVED, actor,
462 pattern_id=pid, metadata={"escalation_id": rec.escalation_id}
463 )
464
465 # 5. now 0 open, 1 resolved
466 assert list_escalations(repo, status=EscalationStatus.OPEN) == []
467 resolved = list_escalations(repo, status=EscalationStatus.RESOLVED)
468 assert len(resolved) == 1
469
470 # 6. audit log has both events
471 entries = list_audit(repo)
472 event_types = {e["event_type"] for e in entries}
473 assert AuditEventType.ESCALATION_RECORDED in event_types
474 assert AuditEventType.ESCALATION_RESOLVED in event_types
475
476 def test_multiple_patterns_independent_escalations(self, repo: pathlib.Path) -> None:
477 pids = [_fake_id(f"track{i}.mid") for i in range(5)]
478 recs = []
479 for pid in pids:
480 rec = _make_record(pid, "no match")
481 record_escalation(repo, rec)
482 recs.append(rec)
483
484 assert len(list_escalations(repo)) == 5
485
486 # resolve two of them
487 for rec in recs[:2]:
488 resolve_escalation(repo, rec.escalation_id, _fake_id("res"),
489 AgentProvenance.human(), _utc_now())
490
491 assert len(list_escalations(repo, status=EscalationStatus.OPEN)) == 3
492 assert len(list_escalations(repo, status=EscalationStatus.RESOLVED)) == 2
493
494
495 # ===========================================================================
496 # Tier IV — Stress
497 # ===========================================================================
498
499
500 class TestStress:
501 """IV: 200 escalations; concurrent record idempotency."""
502
503 def test_list_200_escalations(self, repo: pathlib.Path) -> None:
504 pids = [_fake_id(f"stress-{i}") for i in range(200)]
505 for pid in pids:
506 record_escalation(repo, _make_record(pid, "stress"))
507 results = list_escalations(repo)
508 assert len(results) == 200
509
510 def test_concurrent_record_same_id_idempotent(self, repo: pathlib.Path) -> None:
511 """Concurrent writes of the same escalation_id must not corrupt data."""
512 rec = _make_record()
513 n = 20
514
515 def _write() -> bool:
516 return record_escalation(repo, rec)
517
518 with concurrent.futures.ThreadPoolExecutor(max_workers=n) as ex:
519 futures = [ex.submit(_write) for _ in range(n)]
520 results = [f.result() for f in futures]
521
522 # File must exist and be valid JSON
523 dest = escalations_dir(repo) / f"{rec.escalation_id}.json"
524 assert dest.exists()
525 data = json.loads(dest.read_text())
526 assert data["escalation_id"] == rec.escalation_id
527 # At least one write must have succeeded
528 assert any(results)
529
530
531 # ===========================================================================
532 # Tier V — Data integrity
533 # ===========================================================================
534
535
536 class TestDataIntegrity:
537 """V: JSON round-trip; all fields present; frozen records."""
538
539 def test_json_round_trip_open(self, repo: pathlib.Path) -> None:
540 rec = _make_record()
541 record_escalation(repo, rec)
542 loaded = load_escalation(repo, rec.escalation_id)
543 assert loaded is not None
544 assert loaded.escalation_id == rec.escalation_id
545 assert loaded.pattern_id == rec.pattern_id
546 assert loaded.reason == rec.reason
547 assert loaded.status == EscalationStatus.OPEN
548 assert loaded.resolved_at is None
549 assert loaded.resolved_by is None
550 assert loaded.resolution_id is None
551
552 def test_json_round_trip_resolved(self, repo: pathlib.Path) -> None:
553 rec = _make_record()
554 record_escalation(repo, rec)
555 res_id = _fake_id("res")
556 actor = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
557 now = _utc_now()
558 resolve_escalation(repo, rec.escalation_id, res_id, actor, now)
559 loaded = load_escalation(repo, rec.escalation_id)
560 assert loaded is not None
561 assert loaded.status == EscalationStatus.RESOLVED
562 assert loaded.resolution_id == res_id
563 assert loaded.resolved_by is not None
564 assert loaded.resolved_by.agent_id == "claude-code"
565 assert loaded.resolved_by.model_id == "claude-sonnet-4-6"
566 assert loaded.resolved_at is not None
567
568 def test_escalation_record_is_frozen(self) -> None:
569 rec = _make_record()
570 assert dataclasses.is_dataclass(rec)
571 with pytest.raises((dataclasses.FrozenInstanceError, AttributeError)):
572 rec.reason = "mutated" # type: ignore[misc]
573
574 def test_no_temp_files_in_escalations_dir(self, repo: pathlib.Path) -> None:
575 for i in range(5):
576 record_escalation(repo, _make_record(_fake_id(f"p{i}"), "r"))
577 esc_dir = escalations_dir(repo)
578 assert list(esc_dir.glob("*.tmp")) == []
579
580 def test_escalated_at_is_utc_aware(self, repo: pathlib.Path) -> None:
581 rec = _make_record()
582 record_escalation(repo, rec)
583 loaded = load_escalation(repo, rec.escalation_id)
584 assert loaded is not None
585 assert loaded.escalated_at.tzinfo is not None
586
587
588 # ===========================================================================
589 # Tier VI — Security
590 # ===========================================================================
591
592
593 class TestSecurity:
594 """VI: ID validation; path traversal rejected; size cap."""
595
596 def test_load_traversal_rejected(self, repo: pathlib.Path) -> None:
597 with pytest.raises(ValueError):
598 load_escalation(repo, "../../etc/passwd")
599
600 def test_resolve_traversal_rejected(self, repo: pathlib.Path) -> None:
601 with pytest.raises(ValueError):
602 resolve_escalation(
603 repo, "../evil", _fake_id("r"),
604 AgentProvenance.human(), _utc_now()
605 )
606
607 def test_load_null_byte_rejected(self, repo: pathlib.Path) -> None:
608 with pytest.raises(ValueError):
609 load_escalation(repo, "a" * 63 + "\x00")
610
611 def test_load_too_short_rejected(self, repo: pathlib.Path) -> None:
612 with pytest.raises(ValueError):
613 load_escalation(repo, "abc")
614
615 def test_load_uppercase_hex_rejected(self, repo: pathlib.Path) -> None:
616 with pytest.raises(ValueError):
617 load_escalation(repo, "A" * 64)
618
619 def test_list_skips_non_json_files(self, repo: pathlib.Path) -> None:
620 rec = _make_record()
621 record_escalation(repo, rec)
622 # Plant a non-JSON file
623 esc_dir = escalations_dir(repo)
624 (esc_dir / "not-a-real-file.txt").write_text("noise")
625 results = list_escalations(repo)
626 assert len(results) == 1
627
628 def test_oversized_escalation_file_skipped(self, repo: pathlib.Path) -> None:
629 """Files exceeding _MAX_ESCALATION_BYTES must be silently skipped."""
630 rec = _make_record()
631 record_escalation(repo, rec)
632 dest = escalations_dir(repo) / f"{rec.escalation_id}.json"
633 # Bloat the file beyond 16 KiB
634 bloated = dest.read_text() + " " * 20_000
635 dest.write_text(bloated)
636 results = list_escalations(repo)
637 assert results == []
638
639 def test_symlinks_in_escalations_dir_skipped(self, repo: pathlib.Path) -> None:
640 rec = _make_record()
641 record_escalation(repo, rec)
642 esc_dir = escalations_dir(repo)
643 link = esc_dir / f"{'b' * 64}.json"
644 link.symlink_to(esc_dir / f"{rec.escalation_id}.json")
645 results = list_escalations(repo)
646 assert len(results) == 1 # only real file
647
648
649 # ===========================================================================
650 # Tier VII — Performance
651 # ===========================================================================
652
653
654 class TestPerformance:
655 """VII: all ops complete within 50 ms."""
656
657 def test_record_escalation_under_50ms(self, repo: pathlib.Path) -> None:
658 rec = _make_record()
659 start = time.monotonic()
660 record_escalation(repo, rec)
661 elapsed = (time.monotonic() - start) * 1000
662 assert elapsed < 50, f"record_escalation took {elapsed:.1f}ms"
663
664 def test_load_escalation_under_50ms(self, repo: pathlib.Path) -> None:
665 rec = _make_record()
666 record_escalation(repo, rec)
667 start = time.monotonic()
668 load_escalation(repo, rec.escalation_id)
669 elapsed = (time.monotonic() - start) * 1000
670 assert elapsed < 50, f"load_escalation took {elapsed:.1f}ms"
671
672 def test_resolve_escalation_under_50ms(self, repo: pathlib.Path) -> None:
673 rec = _make_record()
674 record_escalation(repo, rec)
675 start = time.monotonic()
676 resolve_escalation(repo, rec.escalation_id, _fake_id("r"),
677 AgentProvenance.human(), _utc_now())
678 elapsed = (time.monotonic() - start) * 1000
679 assert elapsed < 50, f"resolve_escalation took {elapsed:.1f}ms"
680
681 def test_list_100_escalations_under_50ms(self, repo: pathlib.Path) -> None:
682 for i in range(100):
683 record_escalation(repo, _make_record(_fake_id(f"p{i}"), "r"))
684 start = time.monotonic()
685 list_escalations(repo)
686 elapsed = (time.monotonic() - start) * 1000
687 assert elapsed < 50, f"list_escalations(100) took {elapsed:.1f}ms"
688
689 def test_compute_escalation_id_under_1ms(self) -> None:
690 pid = _fake_id("p")
691 start = time.monotonic()
692 compute_escalation_id(pid, "reason")
693 elapsed = (time.monotonic() - start) * 1000
694 assert elapsed < 1, f"compute_escalation_id took {elapsed:.2f}ms"
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 142 days ago