coulomb/phase-memory
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Markitect bridge and activation quality

Browse Source
This commit is contained in:
Bernd Worsch
2026-05-18 20:04:40 +02:00
parent 8089a7c8fa
commit 1efb7d4c13
19 changed files with 871 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
src/phase_memory/__init__.py
View File

@@ -1,6 +1,14 @@
"""Profile-driven memory phase planning."""
from .activation import plan_activation
from .bridge import (
MARKITECT_PACKAGE_REQUEST_SCHEMA,
MARKITECT_PACKAGE_RESPONSE_SCHEMA,
LocalMarkitectValidator,
OptionalMarkitectValidator,
package_request_from_selection,
package_response_envelope,
)
from .contracts import graph_from_markitect, profile_from_markitect
from .lifecycle import (
plan_compaction,
@@ -30,6 +38,13 @@ from .models import (
)
from .paths import abandon_path, branch_path, compact_path, create_path, merge_path, path_event
from .policy import POLICY_OPERATION_POINTS, MemoryOperation, make_review_record
from .retrieval import (
WordCountTokenEstimator,
activation_quality_report,
plan_neighborhood_activation,
retrieve_graph_neighborhood,
select_event_path,
)
from .planner import plan_profile_execution
from .runtime import PhaseMemoryRuntime
@@ -55,6 +70,10 @@ __all__ = [
"PhaseMemoryRuntime",
"POLICY_OPERATION_POINTS",
"MemoryOperation",
"MARKITECT_PACKAGE_REQUEST_SCHEMA",
"MARKITECT_PACKAGE_RESPONSE_SCHEMA",
"LocalMarkitectValidator",
"OptionalMarkitectValidator",
"abandon_path",
"branch_path",
"compact_path",
@@ -70,6 +89,13 @@ __all__ = [
"plan_retention",
"profile_from_markitect",
"path_event",
"package_request_from_selection",
"package_response_envelope",
"WordCountTokenEstimator",
"activation_quality_report",
"plan_neighborhood_activation",
"retrieve_graph_neighborhood",
"select_event_path",
]
__version__ = "0.1.0"

11
src/phase_memory/activation.py
View File

@@ -17,6 +17,7 @@ def plan_activation(
) -> ActivationPlan:
selected: list[str] = []
omitted: list[dict[str, object]] = []
selected_items: dict[str, dict[str, object]] = {}
token_estimate = 0
ordered_nodes = _ordered_nodes(graph, priority_node_ids)
@@ -29,6 +30,15 @@ def plan_activation(
omitted.append({"id": node.node_id, "reason": "max_tokens", "tokens": node_tokens})
continue
selected.append(node.node_id)
selected_items[node.node_id] = {
"source_spans": list(node.source_spans),
"provenance": list(node.provenance),
"confidence": node.confidence,
"freshness": dict(node.freshness),
"namespace": dict(node.namespace),
"policy": dict(node.policy),
"reason_selected": "priority" if node.node_id in priority_node_ids else "budget_order",
}
token_estimate += node_tokens
selected_event_ids: tuple[str, ...] = ()
@@ -54,6 +64,7 @@ def plan_activation(
"max_items": max_items,
"max_tokens": max_tokens,
"omitted": omitted,
"selected_items": selected_items,
},
}

97
src/phase_memory/bridge.py Normal file
View File

@@ -0,0 +1,97 @@
"""Markitect package bridge helpers."""
from __future__ import annotations
from typing import Any, Protocol
from .contracts import ContractIngressResult, graph_from_markitect, profile_from_markitect, selection_from_markitect
from .models import Diagnostic
from .utils import stable_digest
MARKITECT_PACKAGE_REQUEST_SCHEMA = "phase_memory.markitect.package_request.v1"
MARKITECT_PACKAGE_RESPONSE_SCHEMA = "phase_memory.markitect.package_response.v1"
class MarkitectValidator(Protocol):
def validate_profile(self, data: dict[str, Any]) -> ContractIngressResult: ...
def validate_graph(self, data: dict[str, Any]) -> ContractIngressResult: ...
def validate_selection(self, data: dict[str, Any]) -> ContractIngressResult: ...
class LocalMarkitectValidator:
"""Dependency-light validation boundary using local ingress diagnostics."""
def validate_profile(self, data: dict[str, Any]) -> ContractIngressResult:
return profile_from_markitect(data)
def validate_graph(self, data: dict[str, Any]) -> ContractIngressResult:
return graph_from_markitect(data)
def validate_selection(self, data: dict[str, Any]) -> ContractIngressResult:
return selection_from_markitect(data)
class OptionalMarkitectValidator:
"""Adapter shell for optional Markitect-owned validation."""
def __init__(self, delegate: Any | None = None) -> None:
self.delegate = delegate
self.local = LocalMarkitectValidator()
def validate_profile(self, data: dict[str, Any]) -> ContractIngressResult:
if self.delegate and hasattr(self.delegate, "validate_memory_profile"):
return self.delegate.validate_memory_profile(data)
return self.local.validate_profile(data)
def validate_graph(self, data: dict[str, Any]) -> ContractIngressResult:
if self.delegate and hasattr(self.delegate, "validate_memory_graph"):
return self.delegate.validate_memory_graph(data)
return self.local.validate_graph(data)
def validate_selection(self, data: dict[str, Any]) -> ContractIngressResult:
if self.delegate and hasattr(self.delegate, "validate_memory_selection"):
return self.delegate.validate_memory_selection(data)
return self.local.validate_selection(data)
def package_request_from_selection(
selection: dict[str, Any],
*,
compiler: str,
diagnostics: tuple[Diagnostic, ...] = (),
) -> dict[str, Any]:
metadata = dict(selection.get("metadata") or {})
request_id = f"markitect-package-request:{stable_digest([selection, compiler])}"
return {
"schema_version": MARKITECT_PACKAGE_REQUEST_SCHEMA,
"id": request_id,
"selection_id": selection.get("id"),
"graph_id": selection.get("graph"),
"profile_id": selection.get("profile"),
"selected_nodes": list(selection.get("nodes", ())),
"selected_events": list(selection.get("events", ())),
"budget": {
"max_items": metadata.get("max_items"),
"max_tokens": metadata.get("max_tokens"),
"token_estimate": metadata.get("token_estimate"),
},
"policy": metadata.get("policy", {}),
"provenance": {
"selected_items": metadata.get("selected_items", {}),
"planned_by": metadata.get("planned_by"),
},
"compiler": compiler,
"compiler_diagnostics": [diagnostic.to_dict() for diagnostic in diagnostics],
"selection": dict(selection),
"dry_run": True,
}
def package_response_envelope(response: dict[str, Any], *, request_id: str) -> dict[str, Any]:
return {
"schema_version": MARKITECT_PACKAGE_RESPONSE_SCHEMA,
"request_id": request_id,
"package_ref": response.get("package_id") or response.get("package_ref") or "",
"compiler_diagnostics": list(response.get("diagnostics", ())),
"response": dict(response),
}

182
src/phase_memory/retrieval.py Normal file
View File

@@ -0,0 +1,182 @@
"""Deterministic retrieval and activation quality helpers."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Protocol
from .activation import plan_activation
from .models import ActivationPlan, MemoryEvent, MemoryGraph, MemoryNode, MemoryPath, MemoryPathState
class TokenEstimator(Protocol):
def estimate_text(self, text: str) -> int: ...
def estimate_node(self, node: MemoryNode) -> int: ...
def estimate_event(self, event: MemoryEvent) -> int: ...
class WordCountTokenEstimator:
def estimate_text(self, text: str) -> int:
return max(1, len(text.split()))
def estimate_node(self, node: MemoryNode) -> int:
return self.estimate_text(node.text or node.kind)
def estimate_event(self, event: MemoryEvent) -> int:
return self.estimate_text(" ".join([event.kind, *event.package_refs, *event.activation_refs]))
@dataclass(frozen=True)
class RetrievalCandidate:
node_id: str
distance: int
score: int
reasons: tuple[str, ...]
def to_dict(self) -> dict:
return {
"node_id": self.node_id,
"distance": self.distance,
"score": self.score,
"reasons": list(self.reasons),
}
def retrieve_graph_neighborhood(
graph: MemoryGraph,
*,
seed_node_ids: tuple[str, ...],
max_hops: int = 1,
edge_kinds: tuple[str, ...] = (),
direction: str = "both",
phases: tuple[str, ...] = (),
kinds: tuple[str, ...] = (),
) -> tuple[RetrievalCandidate, ...]:
distances = {node_id: 0 for node_id in seed_node_ids if node_id in graph.node_ids}
frontier = list(distances)
for hop in range(1, max_hops + 1):
next_frontier: list[str] = []
for node_id in frontier:
for edge in graph.edges:
if edge_kinds and edge.kind not in edge_kinds:
continue
neighbors: list[str] = []
if direction in {"out", "both"} and edge.source == node_id:
neighbors.append(edge.target)
if direction in {"in", "both"} and edge.target == node_id:
neighbors.append(edge.source)
for neighbor in neighbors:
if neighbor not in distances:
distances[neighbor] = hop
next_frontier.append(neighbor)
frontier = next_frontier
by_id = graph.node_by_id()
candidates: list[RetrievalCandidate] = []
for node_id, distance in distances.items():
node = by_id[node_id]
if phases and node.phase.value not in phases:
continue
if kinds and node.kind not in kinds:
continue
candidates.append(score_candidate(node, distance=distance, explicit_priority=node_id in seed_node_ids))
return tuple(sorted(candidates, key=lambda item: (-item.score, item.distance, item.node_id)))
def score_candidate(node: MemoryNode, *, distance: int, explicit_priority: bool = False, policy_allowed: bool = True) -> RetrievalCandidate:
score = 100 - (distance * 10)
reasons = [f"graph_distance:{distance}"]
if explicit_priority:
score += 50
reasons.append("explicit_priority")
if node.phase.value in {"stabilized", "rigid"}:
score += 10
reasons.append(f"phase:{node.phase.value}")
if node.lifecycle.value == "active":
score += 5
reasons.append("lifecycle:active")
if node.confidence is not None:
score += int(node.confidence * 10)
reasons.append("confidence")
if node.source_spans:
score += 5
reasons.append("source_backed")
if node.freshness.get("status") == "stale":
score -= 20
reasons.append("freshness:stale")
if not policy_allowed:
score -= 100
reasons.append("policy:denied")
return RetrievalCandidate(node.node_id, distance, score, tuple(reasons))
def select_event_path(
events: tuple[MemoryEvent, ...],
path: MemoryPath,
*,
max_events: int,
include_inactive: bool = False,
) -> tuple[str, ...]:
if path.state != MemoryPathState.ACTIVE and not include_inactive:
return ()
available = {event.event_id for event in events}
return tuple(event_id for event_id in path.event_ids if event_id in available)[:max_events]
def plan_neighborhood_activation(
graph: MemoryGraph,
*,
seed_node_ids: tuple[str, ...],
max_hops: int,
max_items: int,
max_tokens: int,
profile_id: str | None = None,
estimator: TokenEstimator | None = None,
) -> tuple[ActivationPlan, tuple[RetrievalCandidate, ...]]:
estimator = estimator or WordCountTokenEstimator()
candidates = retrieve_graph_neighborhood(graph, seed_node_ids=seed_node_ids, max_hops=max_hops)
priority = tuple(candidate.node_id for candidate in candidates)
plan = plan_activation(
graph,
max_items=max_items,
max_tokens=max_tokens,
profile_id=profile_id,
priority_node_ids=priority,
)
# Preserve the estimator contract for callers without changing the current
# activation plan wire shape.
plan.selection.setdefault("metadata", {})["estimator"] = estimator.__class__.__name__
return plan, candidates
def activation_quality_report(
plan: ActivationPlan,
*,
expected_node_ids: tuple[str, ...] = (),
policy_denied_node_ids: tuple[str, ...] = (),
) -> dict:
selected = set(plan.selected_node_ids)
omitted_ids = {str(item.get("id")) for item in plan.omitted}
expected = set(expected_node_ids)
selected_expected = sorted(selected & expected)
omitted_required = sorted((expected - selected) & omitted_ids)
selected_metadata = plan.selection.get("metadata", {}).get("selected_items", {})
return {
"selected_expected_nodes": selected_expected,
"omitted_required_nodes": omitted_required,
"policy_denied_required_nodes": sorted(expected & set(policy_denied_node_ids)),
"token_budget_utilization": round(plan.token_estimate / plan.max_tokens, 4) if plan.max_tokens else 0,
"stale_item_activation_count": sum(
1 for node_id in plan.selected_node_ids if selected_metadata.get(node_id, {}).get("freshness", {}).get("status") == "stale"
),
"provenance_coverage": _coverage(plan.selected_node_ids, selected_metadata, "provenance"),
"source_span_coverage": _coverage(plan.selected_node_ids, selected_metadata, "source_spans"),
"explanation_coverage": _coverage(plan.selected_node_ids, selected_metadata, "reason_selected"),
}
def _coverage(node_ids: tuple[str, ...], metadata: dict, key: str) -> float:
if not node_ids:
return 0.0
covered = sum(1 for node_id in node_ids if metadata.get(node_id, {}).get(key))
return round(covered / len(node_ids), 4)

14
src/phase_memory/runtime.py
View File

@@ -15,6 +15,7 @@ from .adapters import (
NoopContextPackageCompiler,
RecordingAuditSink,
)
from .bridge import MARKITECT_PACKAGE_REQUEST_SCHEMA, package_request_from_selection, package_response_envelope
from .contracts import ContractIngressResult, graph_from_markitect, profile_from_markitect
from .lifecycle import plan_compaction, plan_refresh, plan_retention
from .models import (
@@ -35,7 +36,7 @@ from .ports import AuditSink, ContextPackageCompiler, MemoryEventLog, MemoryGrap
from .utils import compact_dict, stable_digest, to_plain
RUNTIME_ENVELOPE_SCHEMA = "phase_memory.runtime.envelope.v1"
PACKAGE_REQUEST_SCHEMA = "phase_memory.package_request.v1"
PACKAGE_REQUEST_SCHEMA = MARKITECT_PACKAGE_REQUEST_SCHEMA
@dataclass
@@ -208,7 +209,7 @@ class PhaseMemoryRuntime:
valid=True,
diagnostics=(),
source_ref=source_ref,
data={"package_request": request, "package_response": response},
data={"package_request": request, "package_response": package_response_envelope(response, request_id=request["id"])},
)
def export_graph(self, *, graph_id: str = "local", source_ref: str = "local-store") -> dict[str, Any]:
@@ -301,14 +302,7 @@ class PhaseMemoryRuntime:
)
def package_request(self, selection: dict[str, Any]) -> dict[str, Any]:
request_id = f"package-request:{stable_digest(selection)}"
return {
"schema_version": PACKAGE_REQUEST_SCHEMA,
"id": request_id,
"selection": dict(selection),
"compiler": self.package_compiler.__class__.__name__,
"dry_run": True,
}
return package_request_from_selection(selection, compiler=self.package_compiler.__class__.__name__)
def _contract_envelope(
self,