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

Add profile-driven lifecycle rules

Browse Source
This commit is contained in:
Bernd Worsch
2026-05-18 22:20:14 +02:00
parent 322571c02c
commit 908494b712
12 changed files with 700 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/phase_memory/__init__.py
View File

@@ -11,10 +11,16 @@ from .bridge import (
)
from .contracts import graph_from_markitect, profile_from_markitect
from .lifecycle import (
LifecycleRuleConfig,
PhaseTransitionRule,
RetentionRule,
plan_compaction,
plan_lifecycle_from_profile,
plan_phase_transition,
plan_phase_transitions_from_rules,
plan_refresh,
plan_retention,
plan_retention_from_rules,
)
from .models import (
ActivationPlan,
@@ -69,6 +75,9 @@ __all__ = [
"ReviewDecision",
"ReviewRecord",
"PhaseMemoryRuntime",
"LifecycleRuleConfig",
"PhaseTransitionRule",
"RetentionRule",
"POLICY_OPERATION_POINTS",
"MemoryOperation",
"MARKITECT_PACKAGE_REQUEST_SCHEMA",
@@ -84,10 +93,13 @@ __all__ = [
"make_review_record",
"plan_activation",
"plan_compaction",
"plan_lifecycle_from_profile",
"plan_phase_transition",
"plan_phase_transitions_from_rules",
"plan_profile_execution",
"plan_refresh",
"plan_retention",
"plan_retention_from_rules",
"profile_from_markitect",
"path_event",
"package_request_from_selection",

10
src/phase_memory/cli.py
View File

@@ -41,6 +41,7 @@ def build_parser() -> argparse.ArgumentParser:
lifecycle = graph_subparsers.add_parser("lifecycle", help="Plan graph lifecycle actions")
lifecycle.add_argument("graph", type=Path)
lifecycle.add_argument("--profile", type=Path, help="Derive lifecycle thresholds and rules from a memory profile")
lifecycle.add_argument("--stale-after-days", type=int)
lifecycle.add_argument("--delete-after-days", type=int)
lifecycle.add_argument("--refresh-digest", action="append", default=[], metavar="NODE_ID=DIGEST")
@@ -91,6 +92,15 @@ def _profile_plan(args: argparse.Namespace, runtime: PhaseMemoryRuntime) -> dict
def _graph_lifecycle(args: argparse.Namespace, runtime: PhaseMemoryRuntime) -> dict[str, Any]:
if args.profile:
return runtime.plan_lifecycle_with_profile(
_read_json(args.profile),
_read_json(args.graph),
source_ref=str(args.graph),
profile_source_ref=str(args.profile),
refresh_digests=_parse_digest_args(args.refresh_digest),
compact_node_ids=tuple(args.compact_node),
)
return runtime.plan_lifecycle(
_read_json(args.graph),
source_ref=str(args.graph),

257
src/phase_memory/lifecycle.py
View File

@@ -2,12 +2,104 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
from .models import LifecycleAction, LifecycleActionKind, LifecycleState, MemoryNode, MemoryPhase
from .models import LifecycleAction, LifecycleActionKind, LifecycleState, MemoryGraph, MemoryNode, MemoryPhase, ProfileIntent
from .utils import parse_iso_datetime, stable_digest
@dataclass(frozen=True)
class RetentionRule:
node_kind: str = ""
stale_after_days: int | None = None
delete_after_days: int | None = None
def to_dict(self) -> dict[str, Any]:
data: dict[str, Any] = {}
if self.node_kind:
data["node_kind"] = self.node_kind
if self.stale_after_days is not None:
data["stale_after_days"] = self.stale_after_days
if self.delete_after_days is not None:
data["delete_after_days"] = self.delete_after_days
return data
@dataclass(frozen=True)
class PhaseTransitionRule:
target_phase: MemoryPhase
node_kind: str = ""
from_phase: MemoryPhase | None = None
min_age_days: int | None = None
reason: str = ""
def matches(self, node: MemoryNode, *, now: datetime) -> bool:
if self.node_kind and node.kind != self.node_kind:
return False
if self.from_phase is not None and node.phase != self.from_phase:
return False
if self.min_age_days is not None:
age = _age_days(node, now)
if age is None or age < self.min_age_days:
return False
return node.phase != self.target_phase
def to_dict(self) -> dict[str, Any]:
data: dict[str, Any] = {"target_phase": self.target_phase.value}
if self.node_kind:
data["node_kind"] = self.node_kind
if self.from_phase is not None:
data["from_phase"] = self.from_phase.value
if self.min_age_days is not None:
data["min_age_days"] = self.min_age_days
if self.reason:
data["reason"] = self.reason
return data
@dataclass(frozen=True)
class LifecycleRuleConfig:
retention_default: RetentionRule | None = None
retention_by_kind: dict[str, RetentionRule] | None = None
transition_rules: tuple[PhaseTransitionRule, ...] = ()
refresh_enabled: bool = False
compact_node_ids: tuple[str, ...] = ()
@classmethod
def from_profile(cls, profile: ProfileIntent | dict[str, Any]) -> "LifecycleRuleConfig":
profile_intent = profile if isinstance(profile, ProfileIntent) else ProfileIntent.from_mapping(profile)
retention_default, retention_by_kind = _retention_rules(profile_intent.retention)
return cls(
retention_default=retention_default,
retention_by_kind=retention_by_kind,
transition_rules=_transition_rules(profile_intent),
refresh_enabled=_refresh_enabled(profile_intent.refresh),
compact_node_ids=_string_tuple(
profile_intent.compaction.get("node_ids")
or profile_intent.compaction.get("compact_node_ids")
or ()
),
)
def retention_for(self, node: MemoryNode) -> RetentionRule | None:
by_kind = self.retention_by_kind or {}
return by_kind.get(node.kind) or self.retention_default
def to_dict(self) -> dict[str, Any]:
return {
"retention_default": self.retention_default.to_dict() if self.retention_default else None,
"retention_by_kind": {
key: value.to_dict()
for key, value in sorted((self.retention_by_kind or {}).items())
},
"transition_rules": [rule.to_dict() for rule in self.transition_rules],
"refresh_enabled": self.refresh_enabled,
"compact_node_ids": list(self.compact_node_ids),
}
def plan_phase_transition(
node: MemoryNode,
target_phase: MemoryPhase,
@@ -67,6 +159,47 @@ def plan_retention(
return tuple(actions)
def plan_retention_from_rules(
nodes: list[MemoryNode] | tuple[MemoryNode, ...],
config: LifecycleRuleConfig,
*,
now: datetime | None = None,
) -> tuple[LifecycleAction, ...]:
actions: list[LifecycleAction] = []
for node in nodes:
rule = config.retention_for(node)
if rule is None:
continue
actions.extend(
plan_retention(
[node],
stale_after_days=rule.stale_after_days,
delete_after_days=rule.delete_after_days,
now=now,
)
)
return tuple(actions)
def plan_phase_transitions_from_rules(
nodes: list[MemoryNode] | tuple[MemoryNode, ...],
config: LifecycleRuleConfig,
*,
now: datetime | None = None,
) -> tuple[LifecycleAction, ...]:
now = now or datetime.now(timezone.utc)
actions: list[LifecycleAction] = []
seen: set[tuple[str, str]] = set()
for rule in config.transition_rules:
for node in nodes:
key = (node.node_id, rule.target_phase.value)
if key in seen or not rule.matches(node, now=now):
continue
actions.append(plan_phase_transition(node, rule.target_phase, reason=rule.reason))
seen.add(key)
return tuple(actions)
def plan_compaction(
nodes: list[MemoryNode] | tuple[MemoryNode, ...],
*,
@@ -113,6 +246,29 @@ def plan_refresh(
return tuple(actions)
def plan_lifecycle_from_profile(
graph: MemoryGraph,
profile: ProfileIntent | dict[str, Any],
*,
refresh_digests: dict[str, str] | None = None,
compact_node_ids: tuple[str, ...] = (),
now: datetime | None = None,
) -> tuple[LifecycleAction, ...]:
config = LifecycleRuleConfig.from_profile(profile)
actions: list[LifecycleAction] = []
actions.extend(plan_retention_from_rules(graph.nodes, config, now=now))
actions.extend(plan_phase_transitions_from_rules(graph.nodes, config, now=now))
if config.refresh_enabled and refresh_digests:
actions.extend(plan_refresh(graph.nodes, source_digest_by_node_id=refresh_digests))
by_id = graph.node_by_id()
compact_ids = tuple(dict.fromkeys((*config.compact_node_ids, *compact_node_ids)))
compact_nodes = [by_id[node_id] for node_id in compact_ids if node_id in by_id]
if compact_nodes:
actions.append(plan_compaction(compact_nodes))
return tuple(actions)
def _age_days(node: MemoryNode, now: datetime) -> int | None:
updated = parse_iso_datetime(str(node.freshness.get("updated_at") or node.updated_at))
if updated is None:
@@ -125,3 +281,102 @@ def _summary_text(nodes: list[MemoryNode] | tuple[MemoryNode, ...]) -> str:
if not parts:
return "Summary proposed for selected memory nodes."
return " ".join(parts)[:240]
def _retention_rules(data: dict[str, Any]) -> tuple[RetentionRule | None, dict[str, RetentionRule]]:
if not data:
return None, {}
default = _retention_rule("", data) if _is_retention_rule(data) else None
by_kind: dict[str, RetentionRule] = {}
for key, value in data.items():
if not isinstance(value, dict):
continue
rule = _retention_rule("" if key == "default" else str(key), value)
if key == "default":
default = rule
else:
by_kind[str(key)] = rule
if default is None and len(by_kind) == 1:
only = next(iter(by_kind.values()))
default = RetentionRule(stale_after_days=only.stale_after_days, delete_after_days=only.delete_after_days)
return default, by_kind
def _retention_rule(node_kind: str, data: dict[str, Any]) -> RetentionRule:
return RetentionRule(
node_kind=node_kind,
stale_after_days=_optional_int(data.get("stale_after_days")),
delete_after_days=_optional_int(data.get("delete_after_days")),
)
def _is_retention_rule(data: dict[str, Any]) -> bool:
return "stale_after_days" in data or "delete_after_days" in data
def _transition_rules(profile: ProfileIntent) -> tuple[PhaseTransitionRule, ...]:
raw_rules = (
profile.metadata.get("phase_transitions")
or profile.metadata.get("transition_rules")
or _metadata_lifecycle(profile).get("phase_transitions")
or ()
)
rules: list[PhaseTransitionRule] = []
for item in (raw_rules if isinstance(raw_rules, (list, tuple)) else ()):
if not isinstance(item, dict):
continue
target_phase = _phase(item.get("to_phase") or item.get("target_phase"))
if target_phase is None:
continue
when = item.get("when") if isinstance(item.get("when"), dict) else {}
min_age_days = item.get("min_age_days")
if min_age_days is None:
min_age_days = when.get("min_age_days")
rules.append(
PhaseTransitionRule(
target_phase=target_phase,
node_kind=str(item.get("kind") or item.get("node_kind") or ""),
from_phase=_phase(item.get("from_phase")),
min_age_days=_optional_int(min_age_days),
reason=str(item.get("reason") or ""),
)
)
return tuple(rules)
def _metadata_lifecycle(profile: ProfileIntent) -> dict[str, Any]:
lifecycle = profile.metadata.get("lifecycle")
return dict(lifecycle) if isinstance(lifecycle, dict) else {}
def _refresh_enabled(data: dict[str, Any]) -> bool:
if not data:
return False
trigger = str(data.get("trigger") or data.get("mode") or "enabled").lower()
return trigger not in {"disabled", "off", "none", "false"}
def _phase(value: Any) -> MemoryPhase | None:
if value is None:
return None
try:
return MemoryPhase(str(value))
except ValueError:
return None
def _optional_int(value: Any) -> int | None:
if value is None or value == "":
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def _string_tuple(value: Any) -> tuple[str, ...]:
if value is None:
return ()
if isinstance(value, str):
return (value,)
return tuple(str(item) for item in value)

53
src/phase_memory/runtime.py
View File

@@ -17,7 +17,7 @@ from .adapters import (
)
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 .lifecycle import LifecycleRuleConfig, plan_compaction, plan_lifecycle_from_profile, plan_refresh, plan_retention
from .models import (
Diagnostic,
LifecycleAction,
@@ -150,6 +150,57 @@ class PhaseMemoryRuntime:
},
)
def plan_lifecycle_with_profile(
self,
profile_data: dict[str, Any],
graph_data: dict[str, Any],
*,
source_ref: str = "mapping",
profile_source_ref: str = "profile",
refresh_digests: dict[str, str] | None = None,
compact_node_ids: tuple[str, ...] = (),
now: datetime | None = None,
) -> dict[str, Any]:
profile_result = profile_from_markitect(profile_data)
graph_result = graph_from_markitect(graph_data)
diagnostics = list(profile_result.diagnostics) + list(graph_result.diagnostics)
actions: tuple[LifecycleAction, ...] = ()
rule_config = None
if profile_result.valid and graph_result.valid:
profile: ProfileIntent = profile_result.value
graph: MemoryGraph = graph_result.value
rule_config = LifecycleRuleConfig.from_profile(profile)
actions = plan_lifecycle_from_profile(
graph,
profile,
refresh_digests=refresh_digests or {},
compact_node_ids=compact_node_ids,
now=now,
)
return self._envelope(
"graph.lifecycle.plan",
subject_kind="memory_graph",
subject_id=graph_result.subject_id,
valid=profile_result.valid and graph_result.valid,
diagnostics=diagnostics,
source_ref=source_ref,
data={
"graph_id": graph_result.subject_id,
"profile_id": profile_result.subject_id,
"dry_run_actions": [action.to_dict() for action in actions],
"rule_config": rule_config.to_dict() if rule_config else None,
"parameters": compact_dict(
{
"profile_source_ref": profile_source_ref,
"refresh_digests": refresh_digests or {},
"compact_node_ids": list(compact_node_ids),
}
),
},
)
def plan_activation(
self,
data: dict[str, Any],

21
src/phase_memory/service.py
View File

@@ -29,7 +29,7 @@ KONTEXTUAL_DELEGATION_SCHEMA = "phase_memory.kontextual.delegation.v1"
SERVICE_OPERATIONS = {
"profile.plan": {"request": ["profile"], "response": "runtime_envelope"},
"graph.import": {"request": ["graph"], "response": "runtime_envelope"},
"graph.lifecycle.plan": {"request": ["graph", "parameters"], "response": "runtime_envelope"},
"graph.lifecycle.plan": {"request": ["graph", "parameters", "profile?"], "response": "runtime_envelope"},
"lifecycle.apply": {"request": ["actions", "review_record"], "response": "runtime_envelope"},
"graph.activation.plan": {"request": ["graph", "budget"], "response": "runtime_envelope"},
"package.compile": {"request": ["selection"], "response": "runtime_envelope"},
@@ -353,6 +353,25 @@ class LocalServiceRunner:
return self.runtime.plan_profile(payload["profile"], source_ref=payload.get("source_ref", "service"))
if operation == "graph.import":
return self.runtime.import_graph(payload["graph"], source_ref=payload.get("source_ref", "service"))
if operation == "graph.lifecycle.plan":
parameters = payload.get("parameters", {})
if payload.get("profile"):
return self.runtime.plan_lifecycle_with_profile(
payload["profile"],
payload["graph"],
source_ref=payload.get("source_ref", "service"),
profile_source_ref=payload.get("profile_source_ref", "service-profile"),
refresh_digests=dict(parameters.get("refresh_digests") or {}),
compact_node_ids=tuple(parameters.get("compact_node_ids") or ()),
)
return self.runtime.plan_lifecycle(
payload["graph"],
source_ref=payload.get("source_ref", "service"),
stale_after_days=parameters.get("stale_after_days"),
delete_after_days=parameters.get("delete_after_days"),
refresh_digests=dict(parameters.get("refresh_digests") or {}),
compact_node_ids=tuple(parameters.get("compact_node_ids") or ()),
)
if operation == "graph.activation.plan":
budget = payload.get("budget", {})
return self.runtime.plan_activation(