gabriel / muse public
test_harmony_engine.py python
1,010 lines 37.2 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 148 days ago
1 """Tests for muse/core/harmony_engine.py — Phase 3: Resolution Engine.
2
3 Coverage tiers
4 --------------
5 I Unit — EngineStatus, EngineConfig, EngineResult, DefaultPlugin,
6 HarmonyPlugin protocol, find_similar
7 II Integration — all four resolution tiers (policy, exact-replay,
8 semantic, escalate); action/threshold branching
9 III End-to-end — full pipeline via resolve() with audit trail
10 IV Stress — 100-pattern semantic search; many-policy matching
11 V Data integrity— EngineResult fields always populated; JSON round-trip
12 VI Security — crafted inputs don't escape engine; plugin errors contained
13 VII Performance — resolve() <50 ms; find_similar(100 patterns) <500 ms
14 """
15 from __future__ import annotations
16
17 import dataclasses
18 import datetime
19 import hashlib
20 import pathlib
21 import time
22 from typing import Any
23 from unittest import mock
24
25 import pytest
26
27 import muse.core.harmony as h
28 import muse.core.harmony_engine as eng
29 from muse.core.harmony import (
30 AgentProvenance,
31 AuditEventType,
32 ConflictPattern,
33 ConflictType,
34 Policy,
35 PolicyAction,
36 PolicyCondition,
37 PolicyScope,
38 Resolution,
39 ResolutionProposal,
40 ResolutionStrategy,
41 append_audit,
42 best_resolution,
43 blob_fingerprint,
44 compute_pattern_id,
45 compute_resolution_id,
46 list_audit,
47 list_patterns,
48 list_resolutions,
49 record_pattern,
50 save_policy,
51 save_resolution,
52 )
53 from muse.core.harmony_engine import (
54 DefaultPlugin,
55 EngineConfig,
56 EngineResult,
57 EngineStatus,
58 HarmonyPlugin,
59 find_similar,
60 resolve,
61 )
62
63
64 # ---------------------------------------------------------------------------
65 # Shared helpers
66 # ---------------------------------------------------------------------------
67
68
69 def _fake_id(seed: str) -> str:
70 return hashlib.sha256(seed.encode()).hexdigest()
71
72
73 def _utc_now() -> datetime.datetime:
74 return datetime.datetime.now(datetime.timezone.utc)
75
76
77 @pytest.fixture()
78 def repo(tmp_path: pathlib.Path) -> pathlib.Path:
79 (tmp_path / ".muse").mkdir()
80 return tmp_path
81
82
83 def _make_pattern(
84 path: str = "track.mid",
85 domain: str = "midi",
86 conflict_type: str = ConflictType.CONTENT,
87 ours: str = "ours",
88 theirs: str = "theirs",
89 semantic_fp: str | None = None,
90 ) -> ConflictPattern:
91 ours_id = _fake_id(ours)
92 theirs_id = _fake_id(theirs)
93 blob_fp = blob_fingerprint(ours_id, theirs_id)
94 sfp = semantic_fp if semantic_fp is not None else blob_fp
95 pid = compute_pattern_id(path, blob_fp, sfp)
96 return ConflictPattern(
97 pattern_id=pid,
98 path=path,
99 domain=domain,
100 conflict_type=conflict_type,
101 blob_fingerprint=blob_fp,
102 semantic_fingerprint=sfp,
103 ours_id=ours_id,
104 theirs_id=theirs_id,
105 description={},
106 recorded_at=_utc_now(),
107 recorded_by="claude-code",
108 )
109
110
111 def _make_resolution(
112 pattern: ConflictPattern,
113 confidence: float = 0.9,
114 human_verified: bool = False,
115 strategy: str = ResolutionStrategy.MANUAL,
116 outcome_seed: str = "outcome",
117 applied_count: int = 0,
118 ) -> Resolution:
119 prov = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
120 outcome_blob = _fake_id(outcome_seed)
121 resolved_at = _utc_now()
122 rid = compute_resolution_id(pattern.pattern_id, outcome_blob, strategy, prov, resolved_at)
123 return dataclasses.replace(
124 Resolution(
125 resolution_id=rid,
126 pattern_id=pattern.pattern_id,
127 strategy=strategy,
128 policy_id=None,
129 outcome_blob=outcome_blob,
130 resolved_by=prov,
131 human_verified=human_verified,
132 confidence=confidence,
133 rationale="Test resolution",
134 resolved_at=resolved_at,
135 applied_count=applied_count,
136 ),
137 )
138
139
140 def _make_policy(
141 policy_id: str = "default-policy",
142 scope: str = PolicyScope.REPO,
143 action: str = PolicyAction.PREFER_OURS,
144 confidence: float = 0.95,
145 conflict_type: str | None = None,
146 domain: str | None = None,
147 path_pattern: str | None = None,
148 escalate_to: str | None = None,
149 delegate_to: str | None = None,
150 ) -> Policy:
151 return Policy(
152 policy_id=policy_id,
153 description="Test policy",
154 when=PolicyCondition(
155 conflict_type=conflict_type,
156 domain=domain,
157 path_pattern=path_pattern,
158 ),
159 action=action,
160 confidence=confidence,
161 escalate_to=escalate_to,
162 delegate_to=delegate_to,
163 scope=scope,
164 created_at=_utc_now(),
165 created_by="claude-code",
166 )
167
168
169 # ===========================================================================
170 # Tier I — Unit tests
171 # ===========================================================================
172
173
174 class TestEngineStatus:
175 """I: EngineStatus constants are plain strings."""
176
177 def test_applied_is_string(self) -> None:
178 assert isinstance(EngineStatus.APPLIED, str)
179
180 def test_proposed_is_string(self) -> None:
181 assert isinstance(EngineStatus.PROPOSED, str)
182
183 def test_escalated_is_string(self) -> None:
184 assert isinstance(EngineStatus.ESCALATED, str)
185
186 def test_distinct_values(self) -> None:
187 statuses = {EngineStatus.APPLIED, EngineStatus.PROPOSED, EngineStatus.ESCALATED}
188 assert len(statuses) == 3
189
190
191 class TestEngineConfig:
192 """I: EngineConfig defaults and frozen nature."""
193
194 def test_default_auto_apply_threshold(self) -> None:
195 assert EngineConfig().auto_apply_threshold == pytest.approx(0.85)
196
197 def test_default_semantic_threshold(self) -> None:
198 assert EngineConfig().semantic_threshold == pytest.approx(0.70)
199
200 def test_default_max_proposals(self) -> None:
201 assert EngineConfig().max_proposals == 5
202
203 def test_custom_config(self) -> None:
204 cfg = EngineConfig(auto_apply_threshold=0.99, semantic_threshold=0.50, max_proposals=3)
205 assert cfg.auto_apply_threshold == pytest.approx(0.99)
206 assert cfg.semantic_threshold == pytest.approx(0.50)
207 assert cfg.max_proposals == 3
208
209 def test_frozen(self) -> None:
210 cfg = EngineConfig()
211 with pytest.raises(Exception): # FrozenInstanceError
212 cfg.auto_apply_threshold = 0.0 # type: ignore[misc]
213
214
215 class TestEngineResult:
216 """I: EngineResult fields, defaults, and immutability."""
217
218 def test_applied_result(self) -> None:
219 r = EngineResult(
220 status=EngineStatus.APPLIED,
221 pattern_id=_fake_id("p"),
222 applied_resolution_id=_fake_id("r"),
223 )
224 assert r.status == EngineStatus.APPLIED
225 assert r.proposal is None
226 assert r.escalation_reason is None
227
228 def test_proposed_result(self) -> None:
229 prop = ResolutionProposal(
230 pattern_id=_fake_id("p"),
231 strategy=ResolutionStrategy.POLICY,
232 proposed_action=PolicyAction.PREFER_OURS,
233 confidence=0.9,
234 rationale="test",
235 )
236 r = EngineResult(
237 status=EngineStatus.PROPOSED,
238 pattern_id=_fake_id("p"),
239 proposal=prop,
240 )
241 assert r.proposal is prop
242 assert r.applied_resolution_id is None
243
244 def test_escalated_result(self) -> None:
245 r = EngineResult(
246 status=EngineStatus.ESCALATED,
247 pattern_id=_fake_id("p"),
248 escalation_reason="no match found",
249 )
250 assert r.escalation_reason == "no match found"
251
252 def test_frozen(self) -> None:
253 r = EngineResult(status=EngineStatus.ESCALATED, pattern_id=_fake_id("p"))
254 with pytest.raises(Exception):
255 r.status = EngineStatus.APPLIED # type: ignore[misc]
256
257
258 class TestDefaultPlugin:
259 """I: DefaultPlugin — exact-match similarity, no semantic fingerprint."""
260
261 def test_identical_fps_return_1(self) -> None:
262 fp = _fake_id("same")
263 assert DefaultPlugin().similarity(fp, fp) == pytest.approx(1.0)
264
265 def test_different_fps_return_0(self) -> None:
266 assert DefaultPlugin().similarity(_fake_id("a"), _fake_id("b")) == pytest.approx(0.0)
267
268 def test_similarity_commutative(self) -> None:
269 a, b = _fake_id("x"), _fake_id("y")
270 p = DefaultPlugin()
271 assert p.similarity(a, b) == p.similarity(b, a)
272
273 def test_similarity_range(self) -> None:
274 for seed1, seed2 in [("a", "a"), ("a", "b"), ("c", "d")]:
275 s = DefaultPlugin().similarity(_fake_id(seed1), _fake_id(seed2))
276 assert 0.0 <= s <= 1.0
277
278
279 class TestHarmonyPluginProtocol:
280 """I: HarmonyPlugin is a structural Protocol — custom plugins duck-type in."""
281
282 def test_custom_plugin_accepted(self) -> None:
283 class MyPlugin:
284 def similarity(self, fp_a: str, fp_b: str) -> float:
285 return 0.5
286
287 plugin = MyPlugin()
288 # Should not raise — duck-typing is sufficient
289 result = resolve(
290 pathlib.Path("/nonexistent"),
291 _make_pattern(),
292 plugin=plugin,
293 )
294 # Will escalate because repo doesn't exist, but plugin was accepted
295 assert result.status == EngineStatus.ESCALATED
296
297
298 class TestFindSimilar:
299 """I: find_similar returns correctly sorted proposals."""
300
301 def test_find_similar_empty_store(self, repo: pathlib.Path) -> None:
302 pattern = _make_pattern()
303 record_pattern(repo, pattern)
304 results = find_similar(repo, pattern)
305 assert results == []
306
307 def test_find_similar_skips_self(self, repo: pathlib.Path) -> None:
308 """Even if pattern has the same semantic_fp, it shouldn't match itself."""
309 fp = _fake_id("shared-semantic")
310 pattern = _make_pattern(semantic_fp=fp)
311 record_pattern(repo, pattern)
312 res = _make_resolution(pattern)
313 save_resolution(repo, res)
314 results = find_similar(repo, pattern)
315 assert all(p.similar_pattern_id != pattern.pattern_id for p in results)
316
317 def test_find_similar_returns_semantically_matching_pattern(
318 self, repo: pathlib.Path
319 ) -> None:
320 """Two patterns with same semantic_fp but different paths should match."""
321 shared_fp = _fake_id("shared-semantic")
322
323 source = _make_pattern(path="source.mid", ours="so", theirs="st", semantic_fp=shared_fp)
324 target = _make_pattern(path="target.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
325 record_pattern(repo, source)
326 record_pattern(repo, target)
327
328 # Give source a resolution — target has none
329 res = _make_resolution(source, confidence=0.88)
330 save_resolution(repo, res)
331
332 proposals = find_similar(repo, target)
333 assert len(proposals) >= 1
334 assert proposals[0].similar_pattern_id == source.pattern_id
335 assert proposals[0].similarity == pytest.approx(1.0)
336
337 def test_find_similar_sorted_by_confidence_desc(self, repo: pathlib.Path) -> None:
338 shared_fp = _fake_id("shared-fp")
339 target = _make_pattern(path="t.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
340 record_pattern(repo, target)
341
342 # Two sources with different confidence
343 for i, conf in enumerate([0.5, 0.9]):
344 src = _make_pattern(
345 path=f"src{i}.mid",
346 ours=f"o{i}",
347 theirs=f"t{i}",
348 semantic_fp=shared_fp,
349 )
350 record_pattern(repo, src)
351 r = _make_resolution(src, confidence=conf, outcome_seed=f"out{i}")
352 save_resolution(repo, r)
353
354 proposals = find_similar(repo, target)
355 assert proposals[0].confidence >= proposals[-1].confidence
356
357 def test_find_similar_respects_max_proposals(self, repo: pathlib.Path) -> None:
358 shared_fp = _fake_id("common")
359 target = _make_pattern(path="t.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
360 record_pattern(repo, target)
361
362 for i in range(10):
363 src = _make_pattern(
364 path=f"s{i}.mid", ours=f"o{i}", theirs=f"t{i}", semantic_fp=shared_fp
365 )
366 record_pattern(repo, src)
367 r = _make_resolution(src, confidence=0.7 + i * 0.02, outcome_seed=f"o{i}")
368 save_resolution(repo, r)
369
370 cfg = EngineConfig(max_proposals=3)
371 proposals = find_similar(repo, target, config=cfg)
372 assert len(proposals) <= 3
373
374 def test_find_similar_below_threshold_excluded(self, repo: pathlib.Path) -> None:
375 """DefaultPlugin returns 0.0 for different fingerprints → excluded."""
376 target = _make_pattern(path="t.mid", ours="to", theirs="tt")
377 source = _make_pattern(path="s.mid", ours="so", theirs="st") # different semantic_fp
378 record_pattern(repo, target)
379 record_pattern(repo, source)
380 r = _make_resolution(source)
381 save_resolution(repo, r)
382
383 # With DefaultPlugin, similarity = 0 for different fps → below any threshold
384 proposals = find_similar(repo, target)
385 assert proposals == []
386
387
388 # ===========================================================================
389 # Tier II — Integration: four resolution tiers
390 # ===========================================================================
391
392
393 class TestTierPolicy:
394 """II: Tier 1 — Policy match."""
395
396 def test_policy_prefer_ours_above_threshold_returns_applied(
397 self, repo: pathlib.Path
398 ) -> None:
399 policy = _make_policy(confidence=0.95, action=PolicyAction.PREFER_OURS)
400 save_policy(repo, policy)
401
402 pattern = _make_pattern()
403 record_pattern(repo, pattern)
404
405 result = resolve(repo, pattern)
406 assert result.status == EngineStatus.APPLIED
407 assert result.proposal is not None
408 assert result.proposal.strategy == ResolutionStrategy.POLICY
409 assert result.proposal.proposed_action == PolicyAction.PREFER_OURS
410 assert result.proposal.policy_id == policy.policy_id
411
412 def test_policy_prefer_theirs_above_threshold_returns_applied(
413 self, repo: pathlib.Path
414 ) -> None:
415 policy = _make_policy(confidence=0.90, action=PolicyAction.PREFER_THEIRS)
416 save_policy(repo, policy)
417 pattern = _make_pattern()
418 record_pattern(repo, pattern)
419
420 result = resolve(repo, pattern)
421 assert result.status == EngineStatus.APPLIED
422 assert result.proposal.proposed_action == PolicyAction.PREFER_THEIRS
423
424 def test_policy_below_threshold_returns_proposed(self, repo: pathlib.Path) -> None:
425 policy = _make_policy(confidence=0.60, action=PolicyAction.PREFER_OURS)
426 save_policy(repo, policy)
427 pattern = _make_pattern()
428 record_pattern(repo, pattern)
429
430 cfg = EngineConfig(auto_apply_threshold=0.85)
431 result = resolve(repo, pattern, config=cfg)
432 assert result.status == EngineStatus.PROPOSED
433 assert result.proposal.requires_confirmation is True
434
435 def test_policy_escalate_action_returns_escalated(self, repo: pathlib.Path) -> None:
436 policy = _make_policy(
437 confidence=1.0,
438 action=PolicyAction.ESCALATE,
439 escalate_to="human",
440 )
441 save_policy(repo, policy)
442 pattern = _make_pattern()
443 record_pattern(repo, pattern)
444
445 result = resolve(repo, pattern)
446 assert result.status == EngineStatus.ESCALATED
447 assert result.escalation_reason is not None
448 assert "human" in result.escalation_reason.lower() or "escalat" in result.escalation_reason.lower()
449
450 def test_policy_require_human_returns_escalated(self, repo: pathlib.Path) -> None:
451 policy = _make_policy(confidence=1.0, action=PolicyAction.REQUIRE_HUMAN)
452 save_policy(repo, policy)
453 pattern = _make_pattern()
454 record_pattern(repo, pattern)
455
456 result = resolve(repo, pattern)
457 assert result.status == EngineStatus.ESCALATED
458
459 def test_policy_delegate_returns_escalated(self, repo: pathlib.Path) -> None:
460 policy = _make_policy(
461 confidence=1.0,
462 action=PolicyAction.DELEGATE,
463 delegate_to="harmony-specialist",
464 )
465 save_policy(repo, policy)
466 pattern = _make_pattern()
467 record_pattern(repo, pattern)
468
469 result = resolve(repo, pattern)
470 assert result.status == EngineStatus.ESCALATED
471 assert "harmony-specialist" in (result.escalation_reason or "")
472
473 def test_policy_domain_filter_does_not_fire_for_wrong_domain(
474 self, repo: pathlib.Path
475 ) -> None:
476 policy = _make_policy(domain="code", action=PolicyAction.PREFER_OURS, confidence=1.0)
477 save_policy(repo, policy)
478 pattern = _make_pattern(domain="midi")
479 record_pattern(repo, pattern)
480
481 # Should fall through to escalate (no resolution)
482 result = resolve(repo, pattern)
483 assert result.status == EngineStatus.ESCALATED
484
485 def test_workspace_policy_fires_before_repo_policy(self, repo: pathlib.Path) -> None:
486 workspace_p = _make_policy(
487 "workspace-p",
488 scope=PolicyScope.WORKSPACE,
489 action=PolicyAction.PREFER_OURS,
490 confidence=0.95,
491 )
492 repo_p = _make_policy(
493 "repo-p",
494 scope=PolicyScope.REPO,
495 action=PolicyAction.PREFER_THEIRS,
496 confidence=0.95,
497 )
498 save_policy(repo, workspace_p)
499 save_policy(repo, repo_p)
500 pattern = _make_pattern()
501 record_pattern(repo, pattern)
502
503 result = resolve(repo, pattern)
504 assert result.status == EngineStatus.APPLIED
505 assert result.proposal.proposed_action == PolicyAction.PREFER_OURS # workspace wins
506
507
508 class TestTierExactReplay:
509 """II: Tier 2 — Exact replay (no policy match)."""
510
511 def test_high_confidence_resolution_auto_applied(self, repo: pathlib.Path) -> None:
512 pattern = _make_pattern()
513 record_pattern(repo, pattern)
514 res = _make_resolution(pattern, confidence=0.90)
515 save_resolution(repo, res)
516
517 result = resolve(repo, pattern)
518 assert result.status == EngineStatus.APPLIED
519 assert result.applied_resolution_id == res.resolution_id
520
521 def test_human_verified_always_auto_applied(self, repo: pathlib.Path) -> None:
522 pattern = _make_pattern()
523 record_pattern(repo, pattern)
524 res = _make_resolution(pattern, confidence=0.50, human_verified=True)
525 save_resolution(repo, res)
526
527 cfg = EngineConfig(auto_apply_threshold=0.85)
528 result = resolve(repo, pattern, config=cfg)
529 assert result.status == EngineStatus.APPLIED
530
531 def test_low_confidence_resolution_proposed(self, repo: pathlib.Path) -> None:
532 pattern = _make_pattern()
533 record_pattern(repo, pattern)
534 res = _make_resolution(pattern, confidence=0.60)
535 save_resolution(repo, res)
536
537 cfg = EngineConfig(auto_apply_threshold=0.85)
538 result = resolve(repo, pattern, config=cfg)
539 assert result.status == EngineStatus.PROPOSED
540 assert result.proposal.strategy == ResolutionStrategy.EXACT_REPLAY
541 assert result.proposal.requires_confirmation is True
542
543 def test_exact_replay_increments_applied_count(self, repo: pathlib.Path) -> None:
544 pattern = _make_pattern()
545 record_pattern(repo, pattern)
546 res = _make_resolution(pattern, confidence=0.90)
547 save_resolution(repo, res)
548
549 resolve(repo, pattern)
550
551 from muse.core.harmony import load_resolution
552 loaded = load_resolution(repo, pattern.pattern_id, res.resolution_id)
553 assert loaded is not None
554 assert loaded.applied_count == 1
555
556 def test_exact_replay_prefers_highest_quality_resolution(
557 self, repo: pathlib.Path
558 ) -> None:
559 pattern = _make_pattern()
560 record_pattern(repo, pattern)
561
562 low = _make_resolution(pattern, confidence=0.70, outcome_seed="low")
563 high = _make_resolution(pattern, confidence=0.95, outcome_seed="high")
564 save_resolution(repo, low)
565 save_resolution(repo, high)
566
567 result = resolve(repo, pattern)
568 assert result.applied_resolution_id == high.resolution_id
569
570
571 class TestTierSemantic:
572 """II: Tier 3 — Semantic match (no policy, no exact replay)."""
573
574 def test_semantic_match_returns_proposed(self, repo: pathlib.Path) -> None:
575 shared_fp = _fake_id("shared-semantic")
576
577 source = _make_pattern(path="s.mid", ours="so", theirs="st", semantic_fp=shared_fp)
578 target = _make_pattern(path="t.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
579 record_pattern(repo, source)
580 record_pattern(repo, target)
581
582 res = _make_resolution(source, confidence=0.85)
583 save_resolution(repo, res)
584
585 result = resolve(repo, target)
586 assert result.status == EngineStatus.PROPOSED
587 assert result.proposal.strategy == ResolutionStrategy.SEMANTIC_PROPOSAL
588 assert result.proposal.similar_pattern_id == source.pattern_id
589 assert result.proposal.similarity == pytest.approx(1.0)
590 assert result.proposal.requires_confirmation is True
591
592 def test_semantic_match_below_threshold_escalates(self, repo: pathlib.Path) -> None:
593 """DefaultPlugin returns 0.0 for non-matching fps → no semantic match."""
594 source = _make_pattern(path="s.mid", ours="so", theirs="st")
595 target = _make_pattern(path="t.mid", ours="to", theirs="tt")
596 record_pattern(repo, source)
597 record_pattern(repo, target)
598 res = _make_resolution(source)
599 save_resolution(repo, res)
600
601 result = resolve(repo, target)
602 assert result.status == EngineStatus.ESCALATED
603
604 def test_custom_plugin_similarity_drives_semantic_match(
605 self, repo: pathlib.Path
606 ) -> None:
607 """A custom plugin returning 0.8 similarity enables semantic matching."""
608
609 class AlwaysMatchPlugin:
610 def similarity(self, fp_a: str, fp_b: str) -> float:
611 return 0.8
612
613 source = _make_pattern(path="s.mid", ours="so", theirs="st")
614 target = _make_pattern(path="t.mid", ours="to", theirs="tt")
615 record_pattern(repo, source)
616 record_pattern(repo, target)
617 res = _make_resolution(source, confidence=0.85)
618 save_resolution(repo, res)
619
620 cfg = EngineConfig(semantic_threshold=0.70)
621 result = resolve(repo, target, config=cfg, plugin=AlwaysMatchPlugin())
622 assert result.status == EngineStatus.PROPOSED
623 assert result.proposal.strategy == ResolutionStrategy.SEMANTIC_PROPOSAL
624
625
626 class TestTierEscalate:
627 """II: Tier 4 — Escalation when no policy/replay/semantic matches."""
628
629 def test_no_policy_no_resolution_escalates(self, repo: pathlib.Path) -> None:
630 pattern = _make_pattern()
631 record_pattern(repo, pattern)
632
633 result = resolve(repo, pattern)
634 assert result.status == EngineStatus.ESCALATED
635 assert result.escalation_reason is not None
636 assert len(result.escalation_reason) > 0
637
638 def test_escalation_has_no_applied_resolution(self, repo: pathlib.Path) -> None:
639 pattern = _make_pattern()
640 record_pattern(repo, pattern)
641
642 result = resolve(repo, pattern)
643 assert result.applied_resolution_id is None
644
645 def test_escalation_has_no_proposal(self, repo: pathlib.Path) -> None:
646 pattern = _make_pattern()
647 record_pattern(repo, pattern)
648
649 result = resolve(repo, pattern)
650 assert result.proposal is None
651
652 def test_unrecorded_pattern_escalates(self, repo: pathlib.Path) -> None:
653 """Pattern not in store → escalate (engine is safe with unknown patterns)."""
654 pattern = _make_pattern()
655 # Not recorded
656 result = resolve(repo, pattern)
657 assert result.status == EngineStatus.ESCALATED
658
659
660 # ===========================================================================
661 # Tier III — End-to-end
662 # ===========================================================================
663
664
665 class TestEndToEnd:
666 """III: Full resolution pipeline with audit trail."""
667
668 def test_policy_applied_writes_audit(self, repo: pathlib.Path) -> None:
669 policy = _make_policy(confidence=0.95, action=PolicyAction.PREFER_OURS)
670 save_policy(repo, policy)
671 pattern = _make_pattern()
672 record_pattern(repo, pattern)
673
674 actor = AgentProvenance.agent("claude-code", "claude-sonnet-4-6")
675 result = resolve(repo, pattern, actor=actor)
676 assert result.status == EngineStatus.APPLIED
677
678 entries = list_audit(repo)
679 event_types = [e["event_type"] for e in entries]
680 assert AuditEventType.RESOLUTION_APPLIED in event_types
681
682 def test_exact_replay_applied_writes_audit(self, repo: pathlib.Path) -> None:
683 pattern = _make_pattern()
684 record_pattern(repo, pattern)
685 res = _make_resolution(pattern, confidence=0.90)
686 save_resolution(repo, res)
687
688 actor = AgentProvenance.human()
689 result = resolve(repo, pattern, actor=actor)
690 assert result.status == EngineStatus.APPLIED
691
692 entries = list_audit(repo)
693 event_types = [e["event_type"] for e in entries]
694 assert AuditEventType.RESOLUTION_APPLIED in event_types
695
696 def test_escalation_writes_audit(self, repo: pathlib.Path) -> None:
697 pattern = _make_pattern()
698 record_pattern(repo, pattern)
699
700 actor = AgentProvenance.human()
701 result = resolve(repo, pattern, actor=actor)
702 assert result.status == EngineStatus.ESCALATED
703
704 entries = list_audit(repo)
705 event_types = [e["event_type"] for e in entries]
706 assert AuditEventType.ESCALATION_RECORDED in event_types
707
708 def test_full_lifecycle_policy_to_replay(self, repo: pathlib.Path) -> None:
709 """First resolution via policy; subsequent calls use exact replay."""
710 policy = _make_policy(confidence=0.95, action=PolicyAction.PREFER_OURS)
711 save_policy(repo, policy)
712
713 pattern = _make_pattern()
714 record_pattern(repo, pattern)
715
716 # First call → policy fires
717 r1 = resolve(repo, pattern)
718 assert r1.status == EngineStatus.APPLIED
719 assert r1.proposal.strategy == ResolutionStrategy.POLICY
720
721 # Save the policy resolution so it can be replayed
722 from muse.core.harmony import compute_resolution_id, save_resolution
723 import datetime
724 actor = AgentProvenance.agent("claude-code")
725 resolved_at = datetime.datetime.now(datetime.timezone.utc)
726 rid = compute_resolution_id(
727 pattern.pattern_id, _fake_id("policy-outcome"),
728 ResolutionStrategy.POLICY, actor, resolved_at,
729 )
730 res = Resolution(
731 resolution_id=rid,
732 pattern_id=pattern.pattern_id,
733 strategy=ResolutionStrategy.POLICY,
734 policy_id=policy.policy_id,
735 outcome_blob=_fake_id("policy-outcome"),
736 resolved_by=actor,
737 human_verified=False,
738 confidence=0.95,
739 rationale="Policy applied",
740 resolved_at=resolved_at,
741 )
742 save_resolution(repo, res)
743
744 # Remove policy — next call should use exact replay
745 from muse.core.harmony import remove_policy
746 remove_policy(repo, policy.policy_id)
747
748 r2 = resolve(repo, pattern)
749 assert r2.status == EngineStatus.APPLIED
750 assert r2.proposal.strategy == ResolutionStrategy.EXACT_REPLAY
751
752 def test_resolve_with_explicit_config(self, repo: pathlib.Path) -> None:
753 pattern = _make_pattern()
754 record_pattern(repo, pattern)
755 res = _make_resolution(pattern, confidence=0.80)
756 save_resolution(repo, res)
757
758 # Low threshold → auto-apply
759 cfg_low = EngineConfig(auto_apply_threshold=0.75)
760 r = resolve(repo, pattern, config=cfg_low)
761 assert r.status == EngineStatus.APPLIED
762
763 # High threshold → propose
764 cfg_high = EngineConfig(auto_apply_threshold=0.95)
765 r2 = resolve(repo, pattern, config=cfg_high)
766 assert r2.status == EngineStatus.PROPOSED
767
768
769 # ===========================================================================
770 # Tier IV — Stress
771 # ===========================================================================
772
773
774 class TestStress:
775 """IV: Engine performance under load."""
776
777 def test_engine_with_100_patterns_completes(self, repo: pathlib.Path) -> None:
778 """Engine must handle 100 patterns in the store without crashing."""
779 shared_fp = _fake_id("stress-shared")
780 target = _make_pattern(path="target.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
781 record_pattern(repo, target)
782
783 for i in range(100):
784 p = _make_pattern(
785 path=f"stress{i}.mid",
786 ours=f"so{i}",
787 theirs=f"st{i}",
788 semantic_fp=shared_fp,
789 )
790 record_pattern(repo, p)
791 r = _make_resolution(p, confidence=0.7, outcome_seed=f"o{i}")
792 save_resolution(repo, r)
793
794 result = resolve(repo, target)
795 # Should get a semantic match from one of the 100
796 assert result.status in (EngineStatus.PROPOSED, EngineStatus.ESCALATED)
797
798 def test_find_similar_100_patterns(self, repo: pathlib.Path) -> None:
799 """find_similar with 100 patterns returns correct results."""
800 shared_fp = _fake_id("bulk-shared")
801 target = _make_pattern(path="bulk-target.mid", ours="bto", theirs="btt", semantic_fp=shared_fp)
802 record_pattern(repo, target)
803
804 for i in range(100):
805 p = _make_pattern(
806 path=f"bulk{i}.mid",
807 ours=f"bso{i}",
808 theirs=f"bst{i}",
809 semantic_fp=shared_fp,
810 )
811 record_pattern(repo, p)
812 r = _make_resolution(p, confidence=0.7 + (i % 10) * 0.02, outcome_seed=f"bo{i}")
813 save_resolution(repo, r)
814
815 cfg = EngineConfig(max_proposals=5)
816 proposals = find_similar(repo, target, config=cfg)
817 assert len(proposals) <= 5
818 # All returned proposals should have similarity ≥ threshold
819 for prop in proposals:
820 assert prop.similarity is not None and prop.similarity >= cfg.semantic_threshold
821
822 def test_many_policy_first_match_wins(self, repo: pathlib.Path) -> None:
823 """With 50 policies, the first matching one wins."""
824 for i in range(50):
825 p = _make_policy(
826 f"policy-{i:02d}",
827 scope=PolicyScope.REPO,
828 action=PolicyAction.PREFER_OURS if i == 0 else PolicyAction.PREFER_THEIRS,
829 confidence=0.95,
830 )
831 save_policy(repo, p)
832
833 pattern = _make_pattern()
834 record_pattern(repo, pattern)
835 result = resolve(repo, pattern)
836 # First policy (alphabetically / insertion order) should win
837 assert result.status == EngineStatus.APPLIED
838
839
840 # ===========================================================================
841 # Tier V — Data integrity
842 # ===========================================================================
843
844
845 class TestDataIntegrity:
846 """V: EngineResult fields always populated; JSON-safe."""
847
848 def test_escalated_result_fields_complete(self, repo: pathlib.Path) -> None:
849 pattern = _make_pattern()
850 record_pattern(repo, pattern)
851 result = resolve(repo, pattern)
852 assert result.status is not None
853 assert result.pattern_id is not None
854 # Optional fields default to None (not absent)
855 assert hasattr(result, "proposal")
856 assert hasattr(result, "applied_resolution_id")
857 assert hasattr(result, "escalation_reason")
858
859 def test_applied_result_has_no_escalation_reason(self, repo: pathlib.Path) -> None:
860 policy = _make_policy(confidence=0.95)
861 save_policy(repo, policy)
862 pattern = _make_pattern()
863 record_pattern(repo, pattern)
864 result = resolve(repo, pattern)
865 assert result.escalation_reason is None
866
867 def test_proposal_confidence_in_range(self, repo: pathlib.Path) -> None:
868 policy = _make_policy(confidence=0.95)
869 save_policy(repo, policy)
870 pattern = _make_pattern()
871 record_pattern(repo, pattern)
872 result = resolve(repo, pattern)
873 assert result.proposal is not None
874 assert 0.0 <= result.proposal.confidence <= 1.0
875
876 def test_engine_result_pattern_id_matches_input(self, repo: pathlib.Path) -> None:
877 pattern = _make_pattern()
878 record_pattern(repo, pattern)
879 result = resolve(repo, pattern)
880 assert result.pattern_id == pattern.pattern_id
881
882 def test_proposal_is_json_serialisable(self, repo: pathlib.Path) -> None:
883 import json
884 policy = _make_policy(confidence=0.95)
885 save_policy(repo, policy)
886 pattern = _make_pattern()
887 record_pattern(repo, pattern)
888 result = resolve(repo, pattern)
889 if result.proposal is not None:
890 d = dataclasses.asdict(result.proposal)
891 json.dumps(d) # must not raise
892
893 def test_engine_result_is_json_serialisable(self, repo: pathlib.Path) -> None:
894 import json
895 pattern = _make_pattern()
896 record_pattern(repo, pattern)
897 result = resolve(repo, pattern)
898 d = dataclasses.asdict(result)
899 json.dumps(d) # must not raise
900
901
902 # ===========================================================================
903 # Tier VI — Security
904 # ===========================================================================
905
906
907 class TestSecurity:
908 """VI: Engine handles adversarial inputs safely."""
909
910 def test_plugin_exception_caught(self, repo: pathlib.Path) -> None:
911 """A crashing plugin must not propagate — engine falls back to escalate."""
912
913 class CrashingPlugin:
914 def similarity(self, fp_a: str, fp_b: str) -> float:
915 raise RuntimeError("Plugin exploded")
916
917 source = _make_pattern(path="s.mid", ours="so", theirs="st")
918 target = _make_pattern(path="t.mid", ours="to", theirs="tt")
919 record_pattern(repo, source)
920 record_pattern(repo, target)
921 res = _make_resolution(source)
922 save_resolution(repo, res)
923
924 result = resolve(repo, target, plugin=CrashingPlugin())
925 # Must not raise; engine falls back
926 assert result.status in (EngineStatus.ESCALATED, EngineStatus.PROPOSED)
927
928 def test_plugin_returning_out_of_range_similarity_clamped(
929 self, repo: pathlib.Path
930 ) -> None:
931 """Similarity > 1.0 or < 0.0 from a buggy plugin must be clamped."""
932
933 class OverflowPlugin:
934 def similarity(self, fp_a: str, fp_b: str) -> float:
935 return 999.0
936
937 shared_fp = _fake_id("shared")
938 source = _make_pattern(path="s.mid", ours="so", theirs="st", semantic_fp=shared_fp)
939 target = _make_pattern(path="t.mid", ours="to", theirs="tt", semantic_fp=shared_fp)
940 record_pattern(repo, source)
941 record_pattern(repo, target)
942 res = _make_resolution(source, confidence=0.85)
943 save_resolution(repo, res)
944
945 result = resolve(repo, target, plugin=OverflowPlugin())
946 if result.proposal is not None and result.proposal.similarity is not None:
947 assert 0.0 <= result.proposal.similarity <= 1.0
948
949 def test_resolve_safe_with_empty_repo(self, repo: pathlib.Path) -> None:
950 """Engine against empty store must not crash."""
951 pattern = _make_pattern()
952 result = resolve(repo, pattern)
953 assert result.status == EngineStatus.ESCALATED
954
955 def test_resolve_safe_with_nonexistent_root(self) -> None:
956 """Engine with nonexistent root must escalate, not crash."""
957 pattern = _make_pattern()
958 result = resolve(pathlib.Path("/nonexistent/repo"), pattern)
959 assert result.status == EngineStatus.ESCALATED
960
961
962 # ===========================================================================
963 # Tier VII — Performance
964 # ===========================================================================
965
966
967 class TestPerformance:
968 """VII: Engine timing assertions."""
969
970 def test_resolve_no_match_under_50ms(self, repo: pathlib.Path) -> None:
971 pattern = _make_pattern()
972 record_pattern(repo, pattern)
973 start = time.monotonic()
974 resolve(repo, pattern)
975 elapsed = (time.monotonic() - start) * 1000
976 assert elapsed < 50, f"resolve (escalate) took {elapsed:.1f}ms"
977
978 def test_resolve_policy_under_50ms(self, repo: pathlib.Path) -> None:
979 policy = _make_policy(confidence=0.95)
980 save_policy(repo, policy)
981 pattern = _make_pattern()
982 record_pattern(repo, pattern)
983 start = time.monotonic()
984 resolve(repo, pattern)
985 elapsed = (time.monotonic() - start) * 1000
986 assert elapsed < 50, f"resolve (policy) took {elapsed:.1f}ms"
987
988 def test_resolve_exact_replay_under_50ms(self, repo: pathlib.Path) -> None:
989 pattern = _make_pattern()
990 record_pattern(repo, pattern)
991 res = _make_resolution(pattern, confidence=0.90)
992 save_resolution(repo, res)
993 start = time.monotonic()
994 resolve(repo, pattern)
995 elapsed = (time.monotonic() - start) * 1000
996 assert elapsed < 50, f"resolve (exact replay) took {elapsed:.1f}ms"
997
998 def test_find_similar_100_patterns_under_500ms(self, repo: pathlib.Path) -> None:
999 shared_fp = _fake_id("perf-shared")
1000 target = _make_pattern(path="perf-target.mid", ours="pto", theirs="ptt", semantic_fp=shared_fp)
1001 record_pattern(repo, target)
1002 for i in range(100):
1003 p = _make_pattern(path=f"p{i}.mid", ours=f"po{i}", theirs=f"pt{i}", semantic_fp=shared_fp)
1004 record_pattern(repo, p)
1005 r = _make_resolution(p, confidence=0.7, outcome_seed=f"perf{i}")
1006 save_resolution(repo, r)
1007 start = time.monotonic()
1008 find_similar(repo, target)
1009 elapsed = (time.monotonic() - start) * 1000
1010 assert elapsed < 500, f"find_similar(100) took {elapsed:.1f}ms"
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 148 days ago