"""JSON TypedDicts and serialisation helpers for the harmony CLI.""" from __future__ import annotations from typing import Any, TypedDict from muse.core.envelope import EnvelopeJson, JsonValue from muse.core.harmony import AuditEvent, ConflictPattern, EscalationRecord, Policy, Resolution from muse.core.validation import sanitize_display class _HarmonyRecordJson(EnvelopeJson): """JSON output for ``muse harmony record --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ pattern_id 64-char hex SHA-256 pattern ID derived from the path, blob fingerprint, and semantic fingerprint. already_existed True when the pattern was already recorded — the existing entry is returned unchanged (idempotent). """ pattern_id: str already_existed: bool class _HarmonyListEntryJson(TypedDict): """One conflict pattern entry in ``muse harmony list --json`` output. Nested inside the ``patterns`` list of :class:`_HarmonyListJson`. Fields ------ pattern_id 64-char hex pattern ID. path Workspace-relative POSIX path of the conflicting file. domain Domain name the pattern belongs to (e.g. ``"midi"``). conflict_type Conflict category (content, structural, metadata, …). resolution_count Number of resolutions saved for this pattern. recorded_at ISO-8601 UTC timestamp when the pattern was first recorded. recorded_by Agent ID or ``"human"`` who recorded the pattern. """ pattern_id: str path: str domain: str conflict_type: str resolution_count: int recorded_at: str recorded_by: str class _HarmonyListJson(EnvelopeJson): """JSON output for ``muse harmony list --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ total Total number of patterns returned (after any domain/type filters). patterns Ordered list of pattern entries; see :class:`_HarmonyListEntryJson`. """ total: int patterns: list[_HarmonyListEntryJson] type _PatternDescription = dict[str, JsonValue] class _HarmonyPatternDetailJson(TypedDict): """Full conflict pattern detail nested inside :class:`_HarmonyShowJson`. Fields ------ pattern_id 64-char hex pattern ID. path Workspace-relative POSIX path of the conflicting file. domain Domain name (e.g. ``"midi"``, ``"code"``). conflict_type Conflict category string. blob_fingerprint SHA-256 of ``sorted(ours_id, theirs_id)`` — exact-replay key. semantic_fingerprint Domain-plugin fingerprint — cross-content replay key. Equals ``blob_fingerprint`` when no plugin is active. ours_id 64-char hex object ID for the ours version. theirs_id 64-char hex object ID for the theirs version. description Domain-specific metadata dict (arbitrary JSON object). recorded_at ISO-8601 UTC timestamp when the pattern was first recorded. recorded_by Agent ID or ``"human"`` who recorded the pattern. """ pattern_id: str path: str domain: str conflict_type: str blob_fingerprint: str semantic_fingerprint: str ours_id: str theirs_id: str description: _PatternDescription recorded_at: str recorded_by: str class _HarmonyResolutionDetailJson(TypedDict): """One resolution record — nested in show, best, and similar output. Nested inside :class:`_HarmonyShowJson` (resolutions list) and :class:`_HarmonyBestJson` (resolution field). Fields ------ resolution_id 64-char hex resolution ID. strategy Resolution strategy: ``"manual"``, ``"exact-replay"``, ``"semantic-proposal"``, or ``"policy"``. confidence Reliability score 0.0–1.0 assigned when the resolution was saved. human_verified True when a human confirmed this resolution is correct. Human-verified resolutions always outrank unverified ones. applied_count Number of times this resolution has been auto-applied. resolved_by Provenance dict — ``{"type": "agent"|"human", "agent_id": str|null, "model_id": str|null}``. resolved_at ISO-8601 UTC timestamp when the resolution was saved. rationale Human-readable explanation for why this resolution was chosen. """ resolution_id: str strategy: str confidence: float human_verified: bool applied_count: int resolved_by: dict[str, str | None] resolved_at: str rationale: str class _HarmonyShowJson(EnvelopeJson): """JSON output for ``muse harmony show --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ pattern Full pattern metadata; see :class:`_HarmonyPatternDetailJson`. resolutions All resolutions for this pattern, sorted by quality (human_verified → confidence → applied_count) descending. """ pattern: _HarmonyPatternDetailJson resolutions: list[_HarmonyResolutionDetailJson] class _HarmonyResolveJson(EnvelopeJson): """JSON output for ``muse harmony resolve --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ resolution_id 64-char hex ID of the saved (or existing) resolution. pattern_id 64-char hex ID of the parent conflict pattern. already_existed True when this (outcome_blob, strategy, actor) triple was already saved — idempotent; the existing ID is returned. """ resolution_id: str pattern_id: str already_existed: bool class _HarmonyBestJson(EnvelopeJson): """JSON output for ``muse harmony best --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ pattern_id 64-char hex ID of the queried pattern. resolution The highest-quality resolution (human_verified → confidence → applied_count), or ``null`` when no resolutions exist yet. """ pattern_id: str resolution: _HarmonyResolutionDetailJson | None class _HarmonyForgetJson(EnvelopeJson): """JSON output for ``muse harmony forget --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ pattern_id 64-char hex ID of the pattern that was (or was not) removed. removed True when the pattern and all its resolutions were deleted. False when the pattern did not exist. """ pattern_id: str removed: bool class _HarmonyScalarJson(EnvelopeJson): """JSON output for ``muse harmony clear --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ removed Total number of conflict patterns deleted from the store. """ removed: int class _HarmonyGcJson(EnvelopeJson): """JSON output for ``muse harmony gc --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ removed Number of stale unresolved patterns deleted. age_days Age threshold used — patterns older than this with no resolution were eligible for removal. """ removed: int age_days: int class _HarmonyPolicyAddJson(EnvelopeJson): """JSON output for ``muse harmony policy-add --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ policy_id URL-safe identifier of the policy that was saved or replaced. action The resolution action the policy fires: ``"prefer-ours"``, ``"prefer-theirs"``, ``"escalate"``, ``"require-human"``, or ``"delegate"``. scope Policy scope: ``"workspace"``, ``"repo"``, ``"domain"``, or ``"file"``. """ policy_id: str action: str scope: str class _HarmonyPolicyEntryJson(TypedDict): """One policy record nested inside :class:`_HarmonyPolicyListJson`. Fields ------ policy_id URL-safe policy identifier. description Human-readable explanation of what this policy does. scope Scope level: ``"workspace"``, ``"repo"``, ``"domain"``, or ``"file"``. action Resolution action fired by this policy. confidence Confidence score 0.0–1.0 assigned to policy-driven resolutions. conflict_type Conflict type filter, or ``null`` (fires for all types). domain Domain filter, or ``null`` (fires for all domains). path_pattern fnmatch glob filter on file paths, or ``null``. created_at ISO-8601 UTC timestamp when the policy was created. created_by Creator attribution: agent ID or ``"human"``. """ policy_id: str description: str scope: str action: str confidence: float conflict_type: str | None domain: str | None path_pattern: str | None created_at: str created_by: str class _HarmonyPolicyListJson(EnvelopeJson): """JSON output for ``muse harmony policy-list --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ total Total number of policies returned. policies Scope-sorted policy list (workspace → repo → domain → file, then by created_at ascending within each scope). """ total: int policies: list[_HarmonyPolicyEntryJson] class _HarmonyPolicyRemoveJson(EnvelopeJson): """JSON output for ``muse harmony policy-remove --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ policy_id URL-safe identifier of the policy that was (or was not) removed. removed True when the policy was deleted; False when it did not exist. """ policy_id: str removed: bool class _HarmonyAuditJson(EnvelopeJson): """JSON output for ``muse harmony audit --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ total Total number of entries returned (capped by ``--limit``). entries Audit log entries, newest first; each is a full AuditEvent dict with event_type, occurred_at, actor, pattern_id, and metadata. """ total: int entries: list[AuditEvent] class _HarmonyEscalateJson(EnvelopeJson): """JSON output for ``muse harmony escalate --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ escalation_id 64-char hex ID of the escalation record (deterministic from pattern_id + reason). pattern_id 64-char hex ID of the conflict pattern being escalated. already_existed True when an open escalation for this (pattern_id, reason) pair already existed — idempotent. """ escalation_id: str pattern_id: str already_existed: bool class _HarmonyEscalationEntryJson(TypedDict): """One escalation record nested inside :class:`_HarmonyEscalationsJson`. Fields ------ escalation_id 64-char hex escalation ID. pattern_id 64-char hex pattern ID that was escalated. reason Human-readable reason for escalation. status ``"open"`` or ``"resolved"``. escalated_at ISO-8601 UTC timestamp when the escalation was recorded. escalated_by Provenance dict for the actor who created the escalation. resolved_at ISO-8601 UTC timestamp when the escalation was closed, or ``null`` if still open. resolved_by Provenance dict for the actor who resolved it, or ``null``. resolution_id 64-char hex resolution ID that closed this escalation, or ``null`` if still open. """ escalation_id: str pattern_id: str reason: str status: str escalated_at: str escalated_by: dict[str, str | None] resolved_at: str | None resolved_by: dict[str, str | None] | None resolution_id: str | None class _HarmonyEscalationsJson(EnvelopeJson): """JSON output for ``muse harmony escalations --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ total Total number of escalation records returned. escalations List of escalation entries; see :class:`_HarmonyEscalationEntryJson`. """ total: int escalations: list[_HarmonyEscalationEntryJson] class _HarmonyResolveEscalationJson(EnvelopeJson): """JSON output for ``muse harmony resolve-escalation --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ escalation_id 64-char hex ID of the escalation that was closed. resolved True when the escalation was successfully transitioned to RESOLVED status; False when the escalation was not found. """ escalation_id: str resolved: bool class _HarmonyProposalJson(TypedDict): """A single resolution proposal from the engine or similar search. Nested inside :class:`_HarmonyEngineJson` (proposal field) and :class:`_HarmonySimilarJson` (proposals list). Fields ------ pattern_id 64-char hex ID of the base pattern being resolved. strategy How this proposal was derived: ``"policy"``, ``"exact-replay"``, or ``"semantic-proposal"``. proposed_action The concrete action to take (e.g. ``"prefer-ours"``). confidence Confidence score 0.0–1.0 for this proposal. rationale Explanation of why this resolution was proposed. policy_id If strategy is ``"policy"``, the ID of the matching policy; ``null`` otherwise. similar_pattern_id If strategy is ``"semantic-proposal"``, the ID of the similar pattern; ``null`` otherwise. similarity Similarity score 0.0–1.0 to the similar pattern, or ``null`` when not a semantic proposal. requires_confirmation True when the engine recommends human confirmation before applying this resolution automatically. """ pattern_id: str strategy: str proposed_action: str confidence: float rationale: str policy_id: str | None similar_pattern_id: str | None similarity: float | None requires_confirmation: bool class _HarmonyEngineJson(EnvelopeJson): """JSON output for ``muse harmony engine --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ status Engine outcome: ``"applied"`` (auto-applied a saved resolution), ``"proposed"`` (returned a proposal for human confirmation), or ``"escalated"`` (no match found). pattern_id 64-char hex ID of the pattern evaluated. proposal The resolution proposal, or ``null`` when the engine applied or escalated without proposing. applied_resolution_id 64-char hex ID of the auto-applied resolution, or ``null`` when status is not ``"applied"``. escalation_reason Explanation of why escalation was chosen, or ``null`` when status is not ``"escalated"``. """ status: str pattern_id: str proposal: _HarmonyProposalJson | None applied_resolution_id: str | None escalation_reason: str | None class _HarmonySimilarJson(EnvelopeJson): """JSON output for ``muse harmony similar --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ pattern_id 64-char hex ID of the base pattern searched against. total Total number of similar-pattern proposals returned. proposals List of resolution proposals from semantically similar patterns; see :class:`_HarmonyProposalJson`. """ pattern_id: str total: int proposals: list[_HarmonyProposalJson] # --------------------------------------------------------------------------- # Serialisation helpers # --------------------------------------------------------------------------- def _pattern_to_list_entry( p: ConflictPattern, resolution_count: int, ) -> _HarmonyListEntryJson: return _HarmonyListEntryJson( pattern_id=p.pattern_id, path=sanitize_display(p.path), domain=sanitize_display(p.domain), conflict_type=p.conflict_type, resolution_count=resolution_count, recorded_at=p.recorded_at.isoformat(), recorded_by=sanitize_display(p.recorded_by), ) def _pattern_to_detail(p: ConflictPattern) -> _HarmonyPatternDetailJson: return _HarmonyPatternDetailJson( pattern_id=p.pattern_id, path=sanitize_display(p.path), domain=sanitize_display(p.domain), conflict_type=p.conflict_type, blob_fingerprint=p.blob_fingerprint, semantic_fingerprint=p.semantic_fingerprint, ours_id=p.ours_id, theirs_id=p.theirs_id, description=p.description, recorded_at=p.recorded_at.isoformat(), recorded_by=sanitize_display(p.recorded_by), ) def _resolution_to_detail(r: Resolution) -> _HarmonyResolutionDetailJson: return _HarmonyResolutionDetailJson( resolution_id=r.resolution_id, strategy=r.strategy, confidence=r.confidence, human_verified=r.human_verified, applied_count=r.applied_count, resolved_by=r.resolved_by.to_dict(), resolved_at=r.resolved_at.isoformat(), rationale=r.rationale, ) def _policy_to_entry(p: Policy) -> _HarmonyPolicyEntryJson: return _HarmonyPolicyEntryJson( policy_id=p.policy_id, description=p.description, scope=p.scope, action=p.action, confidence=p.confidence, conflict_type=p.when.conflict_type, domain=p.when.domain, path_pattern=p.when.path_pattern, created_at=p.created_at.isoformat(), created_by=p.created_by, ) def _escalation_to_entry(rec: EscalationRecord) -> _HarmonyEscalationEntryJson: return _HarmonyEscalationEntryJson( escalation_id=rec.escalation_id, pattern_id=rec.pattern_id, reason=sanitize_display(rec.reason), status=rec.status, escalated_at=rec.escalated_at.isoformat(), escalated_by=rec.escalated_by.to_dict(), resolved_at=rec.resolved_at.isoformat() if rec.resolved_at is not None else None, resolved_by=rec.resolved_by.to_dict() if rec.resolved_by is not None else None, resolution_id=rec.resolution_id, )