gabriel / muse public
fingerprint.py python
186 lines 6.9 KB
Raw
sha256:133f9bcf57a62ec7ebf0cd71138b54a200b15989aea2fb2a6912497e2926aa8a feat(dev-safety): Phase 1 of #185 — audit tool for ambiguou… Sonnet 5 patch 2 days ago
1 """Harmony fingerprinting and time helpers — pure computation, no I/O.
2
3 Single responsibility: compute content-addressed IDs and fingerprints for
4 conflict patterns, resolutions, and escalations. All functions are
5 deterministic and side-effect-free except for the current-time helpers
6 (_now_utc, _parse_dt) and the plugin-dispatch in compute_semantic_fingerprint.
7 """
8
9 from __future__ import annotations
10
11 import datetime
12 import logging
13 import pathlib
14
15 from muse.core.types import JsonValue, content_hash
16
17 from .paths import _SHA256_ID_RE
18 from .types import AgentProvenance
19
20 logger = logging.getLogger(__name__)
21
22
23 def _now_utc() -> datetime.datetime:
24 """Return the current UTC time as a timezone-aware datetime."""
25 return datetime.datetime.now(datetime.timezone.utc)
26
27
28 def _parse_dt(value: JsonValue | None) -> datetime.datetime:
29 """Parse *value* as an ISO 8601 datetime, defaulting to now on failure.
30
31 Always returns a UTC-aware datetime regardless of the timezone embedded
32 in *value* — naive datetimes are assumed to be UTC.
33 """
34 try:
35 dt = datetime.datetime.fromisoformat(str(value))
36 if dt.tzinfo is None:
37 dt = dt.replace(tzinfo=datetime.timezone.utc)
38 return dt
39 except (ValueError, TypeError):
40 return _now_utc()
41
42
43 def blob_fingerprint(ours_id: str, theirs_id: str) -> str:
44 """Return the content fingerprint for a pair of conflicting object IDs.
45
46 Sorts the pair lexicographically before hashing so the result is
47 commutative — resolving A-vs-B produces the same fingerprint as B-vs-A.
48
49 Args:
50 ours_id: ``sha256:`` object ID of the "ours" version.
51 theirs_id: ``sha256:`` object ID of the "theirs" version.
52
53 Returns:
54 ``sha256:`` content-addressed fingerprint (length 71).
55 """
56 lo, hi = sorted((ours_id, theirs_id))
57 return content_hash({"ids": [lo, hi]})
58
59
60 def compute_pattern_id(path: str, blob_fp: str, semantic_fp: str) -> str:
61 """Compute the canonical pattern ID for a conflict.
62
63 Incorporates *path* so that two different files with the same conflicting
64 content produce distinct pattern IDs — the resolution for ``track.mid``
65 and ``drums.mid`` are stored and retrieved independently.
66
67 When a domain plugin provides a richer semantic fingerprint (``semantic_fp
68 != blob_fp``), the semantic fingerprint alone determines the pattern
69 identity. This is the mechanism that enables cross-content replay: two
70 merge conflicts with different blob IDs but the same semantic shape
71 (e.g. same musical phrase transposed by a semitone) map to the same
72 pattern and therefore replay the same saved resolution.
73
74 When no semantic plugin is active, ``semantic_fp == blob_fp``, so the
75 formula degenerates to the classic exact-replay model where two conflicts
76 must have identical blob content to share a pattern.
77
78 Args:
79 path: Workspace-relative POSIX path of the conflicting file.
80 blob_fp: ``sha256:`` blob fingerprint from :func:`blob_fingerprint`.
81 semantic_fp: ``sha256:`` semantic fingerprint (from plugin or == blob_fp).
82
83 Returns:
84 ``sha256:`` content-addressed pattern ID (length 71).
85 """
86 if semantic_fp != blob_fp:
87 # Domain plugin provided a richer fingerprint — use it alone so that
88 # conflicts with different blob IDs but the same semantic shape share
89 # a single pattern and replay the same resolution.
90 return content_hash({"path": path, "semantic_fp": semantic_fp})
91 # No plugin (or plugin deferred to blob fingerprint) — include both so
92 # exact-replay is commutative (blob_fp already is) and stable.
93 return content_hash({"blob_fp": blob_fp, "path": path, "semantic_fp": semantic_fp})
94
95
96 def compute_resolution_id(
97 pattern_id: str,
98 outcome_blob: str,
99 strategy: str,
100 resolved_by: AgentProvenance,
101 resolved_at: datetime.datetime,
102 ) -> str:
103 """Compute a stable content-addressed ID for a resolution.
104
105 The ID is deterministic for the same inputs so duplicate saves are
106 idempotent — calling :func:`save_resolution` twice with the same parameters
107 produces the same file and is a no-op.
108
109 Args:
110 pattern_id: ``sha256:`` pattern ID this resolution belongs to.
111 outcome_blob: ``sha256:`` object ID of the resolved content.
112 strategy: :class:`ResolutionStrategy` constant.
113 resolved_by: Attribution for the resolution.
114 resolved_at: UTC-aware timestamp of the resolution.
115
116 Returns:
117 ``sha256:`` content-addressed resolution ID (length 71).
118 """
119 return content_hash({
120 "actor": resolved_by.agent_id or "human",
121 "outcome_blob": outcome_blob,
122 "pattern_id": pattern_id,
123 "resolved_at": resolved_at.isoformat(),
124 "strategy": strategy,
125 })
126
127
128 def compute_escalation_id(pattern_id: str, reason: str) -> str:
129 """Return a deterministic ``sha256:`` escalation ID.
130
131 The same (pattern_id, reason) pair always produces the same escalation_id,
132 enabling idempotent :func:`record_escalation` calls — recording the same
133 escalation twice returns ``False`` without creating a duplicate file.
134
135 Args:
136 pattern_id: ``sha256:`` ID of the pattern that was escalated.
137 reason: The human-readable escalation reason string.
138
139 Returns:
140 ``sha256:`` content-addressed escalation ID (length 71).
141 """
142 return content_hash({"pattern_id": pattern_id, "reason": reason})
143
144
145 def compute_semantic_fingerprint(
146 path: str,
147 ours_id: str,
148 theirs_id: str,
149 plugin: "MuseDomainPlugin",
150 repo_root: pathlib.Path,
151 ) -> str:
152 """Return the semantic fingerprint for a conflict.
153
154 Uses the plugin's ``conflict_fingerprint()`` method if it implements the
155 :class:`~muse.domain.HarmonyPlugin` sub-protocol; falls back to
156 :func:`blob_fingerprint` otherwise.
157
158 Args:
159 path: Workspace-relative POSIX path of the conflicting file.
160 ours_id: SHA-256 object ID of the "ours" blob.
161 theirs_id: SHA-256 object ID of the "theirs" blob.
162 plugin: Active domain plugin instance.
163 repo_root: Repository root for plugin context.
164
165 Returns:
166 sha256: content-addressed fingerprint.
167 """
168 from muse.domain import HarmonyPlugin as _HarmonyPlugin
169
170 if isinstance(plugin, _HarmonyPlugin):
171 try:
172 fp = plugin.conflict_fingerprint(path, ours_id, theirs_id, repo_root)
173 if fp and _SHA256_ID_RE.match(fp):
174 return fp
175 logger.warning(
176 "⚠️ harmony: plugin conflict_fingerprint returned invalid value %r — "
177 "falling back to blob fingerprint",
178 fp,
179 )
180 except Exception as exc:
181 logger.warning(
182 "⚠️ harmony: plugin conflict_fingerprint raised %s — "
183 "falling back to blob fingerprint",
184 exc,
185 )
186 return blob_fingerprint(ours_id, theirs_id)
File History 1 commit
sha256:133f9bcf57a62ec7ebf0cd71138b54a200b15989aea2fb2a6912497e2926aa8a feat(dev-safety): Phase 1 of #185 — audit tool for ambiguou… Sonnet 5 patch 2 days ago