generated from coulomb/repo-seed
106 lines
3.7 KiB
Python
106 lines
3.7 KiB
Python
"""Activation planning and Markitect selection handoff."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from .models import ActivationPlan, Diagnostic, LifecycleAction, LifecycleActionKind, MemoryGraph
|
|
from .utils import stable_digest
|
|
|
|
|
|
def plan_activation(
|
|
graph: MemoryGraph,
|
|
*,
|
|
max_items: int,
|
|
max_tokens: int,
|
|
profile_id: str | None = None,
|
|
priority_node_ids: tuple[str, ...] = (),
|
|
include_events: bool = True,
|
|
) -> 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)
|
|
for node in ordered_nodes:
|
|
node_tokens = _estimate_tokens(node.text or node.kind)
|
|
if len(selected) >= max_items:
|
|
omitted.append({"id": node.node_id, "reason": "max_items"})
|
|
continue
|
|
if token_estimate + node_tokens > max_tokens:
|
|
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, ...] = ()
|
|
if include_events:
|
|
selected_event_ids = tuple(event.event_id for event in graph.events if event.package_refs or event.activation_refs)
|
|
|
|
plan_id = f"activation:{stable_digest([graph.graph_id, selected, selected_event_ids, max_items, max_tokens])}"
|
|
diagnostics = tuple(
|
|
Diagnostic("info", "activation_omitted_items", "Some nodes were omitted by activation budget.", metadata={"count": len(omitted)})
|
|
for _ in [None]
|
|
if omitted
|
|
)
|
|
selection = {
|
|
"schema_version": "markitect.memory.selection.v1",
|
|
"id": plan_id,
|
|
"graph": graph.graph_id,
|
|
"profile": profile_id,
|
|
"nodes": selected,
|
|
"events": list(selected_event_ids),
|
|
"metadata": {
|
|
"planned_by": "phase-memory",
|
|
"token_estimate": token_estimate,
|
|
"max_items": max_items,
|
|
"max_tokens": max_tokens,
|
|
"omitted": omitted,
|
|
"selected_items": selected_items,
|
|
},
|
|
}
|
|
|
|
return ActivationPlan(
|
|
plan_id=plan_id,
|
|
graph_id=graph.graph_id,
|
|
selected_node_ids=tuple(selected),
|
|
selected_event_ids=selected_event_ids,
|
|
omitted=tuple(omitted),
|
|
token_estimate=token_estimate,
|
|
max_items=max_items,
|
|
max_tokens=max_tokens,
|
|
selection=selection,
|
|
diagnostics=diagnostics,
|
|
)
|
|
|
|
|
|
def activation_action(plan: ActivationPlan) -> LifecycleAction:
|
|
return LifecycleAction(
|
|
LifecycleActionKind.ACTIVATE,
|
|
target_id=plan.plan_id,
|
|
reason="compile planned selection through Markitect context-package boundary",
|
|
metadata={"selection": plan.selection},
|
|
)
|
|
|
|
|
|
def _ordered_nodes(graph: MemoryGraph, priority_node_ids: tuple[str, ...]):
|
|
by_id = graph.node_by_id()
|
|
priority = [by_id[node_id] for node_id in priority_node_ids if node_id in by_id]
|
|
remaining = sorted(
|
|
(node for node in graph.nodes if node.node_id not in set(priority_node_ids)),
|
|
key=lambda node: node.node_id,
|
|
)
|
|
return priority + remaining
|
|
|
|
|
|
def _estimate_tokens(text: str) -> int:
|
|
return max(1, len(text.split()))
|