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

feat(memory): add agent safe APIs

Browse Source
This commit is contained in:
Bernd Worsch
2026-05-15 09:51:18 +02:00
parent 153427ae99
commit a2f924df15
6 changed files with 709 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
docs/memory-graph-runtime.md
View File

@@ -33,6 +33,13 @@ with `markitect-tool`.
- `MemoryRuntimeService.compact_memory()` creates a deterministic summary node,
preserves source spans/provenance, optionally retires source nodes, and appends
a compaction event.
- `MemoryRuntimeService.plan_memory_update()` creates dry-run update plans with
source and policy explanations.
- `MemoryRuntimeService.apply_memory_update()` requires explicit approval for
durable writes when a plan is review-gated.
- `MemoryRuntimeService.export_context_package_inputs()` emits
Markitect-compatible context package input envelopes without invoking the
Markitect compiler.
## Boundary
@@ -48,6 +55,7 @@ with `markitect-tool`.
- append-only event storage
- permission-aware retrieval and context assembly
- retention, refresh, compaction, review gates, and audit behavior
- agent-safe update plans and Markitect-compatible export envelopes
`infospace-bench` should consume these records and Markitect fixtures to measure
retrieval quality, latency, budget pressure, and regression behavior.

14
src/kontextual_engine/__init__.py
View File

@@ -139,12 +139,19 @@ from .services import (
MemoryGraphImportSummary,
MemoryLifecycleNodeUpdate,
MemoryLifecycleResult,
MemoryNodeUpdateInstruction,
MemoryPackageExportRequest,
MemoryPackageExportResult,
MemoryPlannedNodeUpdate,
MemoryQueryRequest,
MemoryRefreshRequest,
MemoryRetrievalItem,
MemoryRetrievalResult,
MemoryRetentionRequest,
MemoryRuntimeService,
MemoryUpdatePlan,
MemoryUpdateRequest,
MemoryUpdateResult,
RelationshipChangeResult,
RelationshipQueryItem,
RelationshipQueryRequest,
@@ -271,6 +278,10 @@ __all__ = [
"MemoryLifecycleNodeUpdate",
"MemoryLifecycleResult",
"MemoryNodeRecord",
"MemoryNodeUpdateInstruction",
"MemoryPackageExportRequest",
"MemoryPackageExportResult",
"MemoryPlannedNodeUpdate",
"MemoryProfileRecord",
"MemoryQueryRequest",
"MemoryRefreshRequest",
@@ -279,6 +290,9 @@ __all__ = [
"MemoryRetentionRequest",
"MemoryRuntimeService",
"MemorySourceSpan",
"MemoryUpdatePlan",
"MemoryUpdateRequest",
"MemoryUpdateResult",
"NormalizedDocument",
"NotFoundError",
"OperationFailure",

14
src/kontextual_engine/services/__init__.py
View File

@@ -12,12 +12,19 @@ from .memory_service import (
MemoryGraphImportSummary,
MemoryLifecycleNodeUpdate,
MemoryLifecycleResult,
MemoryNodeUpdateInstruction,
MemoryPackageExportRequest,
MemoryPackageExportResult,
MemoryPlannedNodeUpdate,
MemoryQueryRequest,
MemoryRefreshRequest,
MemoryRetrievalItem,
MemoryRetrievalResult,
MemoryRetentionRequest,
MemoryRuntimeService,
MemoryUpdatePlan,
MemoryUpdateRequest,
MemoryUpdateResult,
)
from .retrieval_service import (
AssetQueryItem,
@@ -69,12 +76,19 @@ __all__ = [
"MemoryGraphImportSummary",
"MemoryLifecycleNodeUpdate",
"MemoryLifecycleResult",
"MemoryNodeUpdateInstruction",
"MemoryPackageExportRequest",
"MemoryPackageExportResult",
"MemoryPlannedNodeUpdate",
"MemoryQueryRequest",
"MemoryRefreshRequest",
"MemoryRetrievalItem",
"MemoryRetrievalResult",
"MemoryRetentionRequest",
"MemoryRuntimeService",
"MemoryUpdatePlan",
"MemoryUpdateRequest",
"MemoryUpdateResult",
"RelationshipChangeResult",
"RepresentationContentResult",
"RepresentationContentStream",

581
src/kontextual_engine/services/memory_service.py
View File

@@ -276,6 +276,178 @@ class MemoryLifecycleResult:
}
@dataclass(frozen=True)
class MemoryNodeUpdateInstruction:
contract_node_id: str
operation: str = "upsert_node"
kind: str | None = None
text: str | None = None
lifecycle: str | None = None
source_spans: tuple[dict[str, Any], ...] = ()
provenance: tuple[dict[str, Any], ...] = ()
freshness: dict[str, Any] = field(default_factory=dict)
policy: dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
object.__setattr__(self, "source_spans", tuple(dict(item) for item in self.source_spans))
object.__setattr__(self, "provenance", tuple(dict(item) for item in self.provenance))
def to_dict(self) -> dict[str, Any]:
return {
"contract_node_id": self.contract_node_id,
"operation": self.operation,
"kind": self.kind,
"text": self.text,
"lifecycle": self.lifecycle,
"source_spans": [dict(item) for item in self.source_spans],
"provenance": [dict(item) for item in self.provenance],
"freshness": dict(self.freshness),
"policy": dict(self.policy),
"metadata": dict(self.metadata),
}
@dataclass(frozen=True)
class MemoryUpdateRequest:
graph_id: str
instructions: tuple[MemoryNodeUpdateInstruction, ...] = ()
require_review: bool = True
reason: str | None = None
def __post_init__(self) -> None:
object.__setattr__(self, "instructions", tuple(self.instructions))
def to_dict(self) -> dict[str, Any]:
return {
"graph_id": self.graph_id,
"instructions": [instruction.to_dict() for instruction in self.instructions],
"require_review": self.require_review,
"reason": self.reason,
}
@dataclass(frozen=True)
class MemoryPlannedNodeUpdate:
action: str
before: MemoryNodeRecord | None
after: MemoryNodeRecord
source_explanation: tuple[dict[str, Any], ...] = ()
policy_explanation: dict[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
object.__setattr__(self, "source_explanation", tuple(dict(item) for item in self.source_explanation))
def to_dict(self) -> dict[str, Any]:
return {
"action": self.action,
"before": self.before.to_dict() if self.before else None,
"after": self.after.to_dict(),
"source_explanation": [dict(item) for item in self.source_explanation],
"policy_explanation": dict(self.policy_explanation),
}
@dataclass(frozen=True)
class MemoryUpdatePlan:
plan_id: str
request: MemoryUpdateRequest
correlation_id: str
dry_run: bool = True
review_required: bool = True
planned_updates: tuple[MemoryPlannedNodeUpdate, ...] = ()
diagnostics: tuple[Diagnostic, ...] = ()
policy_decision: PolicyDecision | None = None
success: bool = True
def __post_init__(self) -> None:
object.__setattr__(self, "planned_updates", tuple(self.planned_updates))
object.__setattr__(self, "diagnostics", tuple(self.diagnostics))
def to_dict(self) -> dict[str, Any]:
return {
"plan_id": self.plan_id,
"request": self.request.to_dict(),
"correlation_id": self.correlation_id,
"dry_run": self.dry_run,
"review_required": self.review_required,
"success": self.success,
"planned_updates": [update.to_dict() for update in self.planned_updates],
"diagnostics": [diagnostic.to_dict() for diagnostic in self.diagnostics],
"policy_decision": self.policy_decision.to_dict() if self.policy_decision else None,
}
@dataclass(frozen=True)
class MemoryUpdateResult:
plan: MemoryUpdatePlan
applied_nodes: tuple[MemoryNodeRecord, ...] = ()
appended_events: tuple[MemoryEventRecord, ...] = ()
audit_event: AuditEvent | None = None
diagnostics: tuple[Diagnostic, ...] = ()
review_required: bool = False
success: bool = True
def __post_init__(self) -> None:
object.__setattr__(self, "applied_nodes", tuple(self.applied_nodes))
object.__setattr__(self, "appended_events", tuple(self.appended_events))
object.__setattr__(self, "diagnostics", tuple(self.diagnostics))
def to_dict(self) -> dict[str, Any]:
return {
"plan": self.plan.to_dict(),
"success": self.success,
"review_required": self.review_required,
"applied_nodes": [node.to_dict() for node in self.applied_nodes],
"appended_events": [event.to_dict() for event in self.appended_events],
"audit_event": self.audit_event.to_dict() if self.audit_event else None,
"diagnostics": [diagnostic.to_dict() for diagnostic in self.diagnostics],
}
@dataclass(frozen=True)
class MemoryPackageExportRequest:
query: MemoryQueryRequest
title: str
intent: str
namespace: dict[str, Any] = field(default_factory=dict)
budget: dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
return {
"query": self.query.to_dict(),
"title": self.title,
"intent": self.intent,
"namespace": dict(self.namespace),
"budget": dict(self.budget),
}
@dataclass(frozen=True)
class MemoryPackageExportResult:
request: MemoryPackageExportRequest
correlation_id: str
package_input: dict[str, Any] = field(default_factory=dict)
retrieval: MemoryRetrievalResult | None = None
audit_event: AuditEvent | None = None
diagnostics: tuple[Diagnostic, ...] = ()
success: bool = True
def __post_init__(self) -> None:
object.__setattr__(self, "diagnostics", tuple(self.diagnostics))
def to_dict(self) -> dict[str, Any]:
return {
"request": self.request.to_dict(),
"correlation_id": self.correlation_id,
"success": self.success,
"package_input": dict(self.package_input),
"retrieval": self.retrieval.to_dict() if self.retrieval else None,
"audit_event": self.audit_event.to_dict() if self.audit_event else None,
"diagnostics": [diagnostic.to_dict() for diagnostic in self.diagnostics],
}
class MemoryRuntimeService:
def __init__(
self,
@@ -690,6 +862,161 @@ class MemoryRuntimeService:
metadata={"source_nodes": len(nodes), "retired_source_nodes": len(retired_nodes)},
)
def plan_memory_update(
self,
request: MemoryUpdateRequest,
context: OperationContext,
) -> MemoryUpdatePlan:
diagnostics = _validate_update_request(request)
plan_id = new_id("memplan")
decision = self._authorize(
context,
"memory.update.plan",
f"memory-graph:{request.graph_id}",
resource_metadata={"request": request.to_dict()},
)
if diagnostics or not decision.allowed:
if not decision.allowed:
diagnostics.append(_permission_denied_diagnostic(decision))
return MemoryUpdatePlan(
plan_id=plan_id,
request=request,
correlation_id=context.correlation_id,
review_required=request.require_review,
diagnostics=tuple(diagnostics),
policy_decision=decision,
success=False,
)
nodes = self.repository.list_memory_nodes(graph_id=request.graph_id)
graph_context = _graph_context(nodes, request.graph_id)
planned: list[MemoryPlannedNodeUpdate] = []
for instruction in request.instructions:
try:
planned.append(_planned_update_for_instruction(instruction, graph_context, context))
except ValueError as exc:
diagnostics.append(Diagnostic("error", "memory.update.invalid", str(exc)))
return MemoryUpdatePlan(
plan_id=plan_id,
request=request,
correlation_id=context.correlation_id,
review_required=request.require_review,
planned_updates=tuple(planned),
diagnostics=tuple(diagnostics),
policy_decision=decision,
success=not diagnostics,
)
def apply_memory_update(
self,
plan: MemoryUpdatePlan,
context: OperationContext,
*,
review_decision: str | None = None,
) -> MemoryUpdateResult:
if not plan.success:
return MemoryUpdateResult(
plan=plan,
diagnostics=plan.diagnostics,
success=False,
)
if plan.review_required and review_decision != "approved":
diagnostic = Diagnostic(
"warning",
"memory.update.review_required",
"Memory update requires explicit review approval before durable write.",
details={"plan_id": plan.plan_id, "review_decision": review_decision},
)
audit_event = AuditEvent.from_context(
"memory.update.apply",
f"memory-graph:{plan.request.graph_id}",
AuditOutcome.REVIEW_REQUIRED,
context,
details={"plan_id": plan.plan_id, "review_decision": review_decision},
)
return MemoryUpdateResult(
plan=plan,
audit_event=audit_event,
diagnostics=(diagnostic,),
review_required=True,
success=False,
)
decision = self._authorize(
context,
"memory.update.apply",
f"memory-graph:{plan.request.graph_id}",
resource_metadata={"plan": plan.to_dict(), "review_decision": review_decision},
)
if not decision.allowed:
diagnostic = _permission_denied_diagnostic(decision)
audit_event = AuditEvent.from_context(
"memory.update.apply",
f"memory-graph:{plan.request.graph_id}",
AuditOutcome.DENIED,
context,
policy_decision=decision,
details={"plan_id": plan.plan_id},
)
return MemoryUpdateResult(plan=plan, audit_event=audit_event, diagnostics=(diagnostic,), success=False)
for update in plan.planned_updates:
self.repository.save_memory_node(update.after)
event = self._append_lifecycle_event(
"updated",
plan.request.graph_id,
context,
node_updates=[_planned_update_event_payload(update) for update in plan.planned_updates],
metadata={"plan_id": plan.plan_id, "review_decision": review_decision},
)
audit_event = AuditEvent.from_context(
"memory.update.apply",
f"memory-graph:{plan.request.graph_id}",
AuditOutcome.SUCCESS,
context,
policy_decision=decision,
details={"plan_id": plan.plan_id, "applied_nodes": len(plan.planned_updates), "event_id": event.event_id},
)
return MemoryUpdateResult(
plan=plan,
applied_nodes=tuple(update.after for update in plan.planned_updates),
appended_events=(event,),
audit_event=audit_event,
)
def export_context_package_inputs(
self,
request: MemoryPackageExportRequest,
context: OperationContext,
) -> MemoryPackageExportResult:
retrieval = self.query_memory(request.query, context)
if not retrieval.success:
return MemoryPackageExportResult(
request=request,
correlation_id=context.correlation_id,
retrieval=retrieval,
diagnostics=retrieval.diagnostics,
success=False,
)
package_input = _markitect_package_input(request, retrieval)
outcome = AuditOutcome.PARTIAL if retrieval.metadata.get("permission_filtered_count") else AuditOutcome.SUCCESS
audit_event = AuditEvent.from_context(
"memory.package.export",
f"memory-graph:{request.query.graph_id or '*'}",
outcome,
context,
details={
"request": request.to_dict(),
"items": len(package_input.get("items", ())),
"permission_filtered_count": retrieval.metadata.get("permission_filtered_count", 0),
},
)
return MemoryPackageExportResult(
request=request,
correlation_id=context.correlation_id,
package_input=package_input,
retrieval=retrieval,
audit_event=audit_event,
diagnostics=retrieval.diagnostics,
)
def _authorize(
self,
context: OperationContext,
@@ -1003,6 +1330,260 @@ def _deterministic_compaction_summary(nodes: list[MemoryNodeRecord]) -> str:
return "\n".join(lines)
def _validate_update_request(request: MemoryUpdateRequest) -> list[Diagnostic]:
diagnostics = _validate_node_selection_request(request.graph_id, ())
if not request.instructions:
diagnostics.append(Diagnostic("error", "memory.update.empty", "At least one update instruction is required."))
for instruction in request.instructions:
if not instruction.contract_node_id:
diagnostics.append(Diagnostic("error", "memory.update.node_id_missing", "contract_node_id is required."))
if instruction.operation not in {"upsert_node", "transition_lifecycle"}:
diagnostics.append(
Diagnostic(
"error",
"memory.update.operation_invalid",
"operation must be upsert_node or transition_lifecycle.",
details={"operation": instruction.operation},
)
)
if instruction.operation == "transition_lifecycle" and not instruction.lifecycle:
diagnostics.append(
Diagnostic(
"error",
"memory.update.lifecycle_missing",
"transition_lifecycle requires a lifecycle value.",
details={"contract_node_id": instruction.contract_node_id},
)
)
return diagnostics
def _graph_context(nodes: list[MemoryNodeRecord], graph_id: str) -> dict[str, Any]:
nodes_by_contract = {node.contract_node_id: node for node in nodes}
nodes_by_id = {str(node.node_id): node for node in nodes}
first = nodes[0] if nodes else None
return {
"graph_id": graph_id,
"contract_graph_id": first.contract_graph_id if first else graph_id,
"namespace": dict(first.namespace) if first else {},
"nodes_by_contract": nodes_by_contract,
"nodes_by_id": nodes_by_id,
}
def _planned_update_for_instruction(
instruction: MemoryNodeUpdateInstruction,
graph_context: dict[str, Any],
context: OperationContext,
) -> MemoryPlannedNodeUpdate:
existing = (
graph_context["nodes_by_contract"].get(instruction.contract_node_id)
or graph_context["nodes_by_id"].get(instruction.contract_node_id)
)
if instruction.operation == "transition_lifecycle" and existing is None:
raise ValueError(f"Cannot transition unknown memory node `{instruction.contract_node_id}`.")
now = utc_now().isoformat()
source_spans = _instruction_source_spans(instruction, existing)
provenance = _instruction_provenance(instruction, existing, context)
if existing is None:
after = MemoryNodeRecord(
graph_id=graph_context["graph_id"],
contract_graph_id=graph_context["contract_graph_id"],
contract_node_id=instruction.contract_node_id,
kind=instruction.kind or "memory",
text=instruction.text or "",
namespace=dict(graph_context["namespace"]),
source_spans=source_spans,
provenance=provenance,
freshness=dict(instruction.freshness),
policy=dict(instruction.policy),
metadata=_agent_update_metadata(instruction.metadata, context, "create_node", now),
lifecycle=LifecycleState(instruction.lifecycle) if instruction.lifecycle else LifecycleState.ACTIVE,
created_at=now,
updated_at=now,
)
action = "create_node"
else:
metadata = dict(existing.metadata)
metadata.update(instruction.metadata)
metadata = _agent_update_metadata(metadata, context, instruction.operation, now)
freshness = dict(existing.freshness)
freshness.update(instruction.freshness)
policy = dict(existing.policy)
policy.update(instruction.policy)
after = replace(
existing,
kind=instruction.kind or existing.kind,
text=existing.text if instruction.text is None else instruction.text,
source_spans=source_spans,
provenance=provenance,
freshness=freshness,
policy=policy,
metadata=metadata,
lifecycle=LifecycleState(instruction.lifecycle) if instruction.lifecycle else existing.lifecycle,
updated_at=now,
)
action = "transition_lifecycle" if instruction.operation == "transition_lifecycle" else "update_node"
return MemoryPlannedNodeUpdate(
action=action,
before=existing,
after=after,
source_explanation=_source_explanation(after),
policy_explanation={
"policy": dict(after.policy),
"actor_id": context.actor.id,
"correlation_id": context.correlation_id,
"durable_write_requires_apply": True,
},
)
def _instruction_source_spans(
instruction: MemoryNodeUpdateInstruction,
existing: MemoryNodeRecord | None,
) -> tuple[MemorySourceSpan, ...]:
if instruction.source_spans:
return tuple(MemorySourceSpan.from_contract(span) for span in instruction.source_spans)
if existing:
return existing.source_spans
return ()
def _instruction_provenance(
instruction: MemoryNodeUpdateInstruction,
existing: MemoryNodeRecord | None,
context: OperationContext,
) -> tuple[dict[str, Any], ...]:
provenance = [dict(item) for item in existing.provenance] if existing else []
provenance.extend(dict(item) for item in instruction.provenance)
provenance.append(
{
"kind": "agent-memory-update-plan",
"actor_id": context.actor.id,
"correlation_id": context.correlation_id,
"operation": instruction.operation,
}
)
return tuple(provenance)
def _agent_update_metadata(
metadata: dict[str, Any],
context: OperationContext,
action: str,
timestamp: str,
) -> dict[str, Any]:
updated = dict(metadata)
updated["agent_update"] = {
"action": action,
"actor_id": context.actor.id,
"correlation_id": context.correlation_id,
"at": timestamp,
}
return updated
def _source_explanation(node: MemoryNodeRecord) -> tuple[dict[str, Any], ...]:
if node.source_spans:
return tuple(
{
"path": span.path,
"selector": span.selector,
"unit_kind": span.unit_kind,
"engine": span.engine,
}
for span in node.source_spans
)
return (
{
"path": f"memory://{node.graph_id}/nodes/{node.node_id}",
"selector": f"nodes[id={node.contract_node_id}]",
"unit_kind": node.kind,
"engine": "kontextual-memory",
},
)
def _planned_update_event_payload(update: MemoryPlannedNodeUpdate) -> dict[str, Any]:
return {
"action": update.action,
"node_id": update.after.node_id,
"contract_node_id": update.after.contract_node_id,
"before_lifecycle": update.before.lifecycle.value if update.before else "missing",
"after_lifecycle": update.after.lifecycle.value,
"source_explanation": [dict(item) for item in update.source_explanation],
}
def _markitect_package_input(
request: MemoryPackageExportRequest,
retrieval: MemoryRetrievalResult,
) -> dict[str, Any]:
graph_id = request.query.graph_id or "*"
data: dict[str, Any] = {
"schema_version": "markitect.context-package.input.v1",
"title": request.title,
"intent": request.intent,
"namespace": dict(request.namespace),
"budget": dict(request.budget),
"retrieval_recipes": [
{
"kind": "memory-runtime-query",
"query": request.query.to_dict(),
"engine": "kontextual-engine.memory",
"sources": [f"memory://{graph_id}"],
"metadata": {
"correlation_id": retrieval.correlation_id,
"permission_filtered_count": retrieval.metadata.get("permission_filtered_count", 0),
},
}
],
"items": [_markitect_context_item(item) for item in retrieval.items],
"metadata": {
"runtime": "kontextual-engine",
"export_kind": "markitect-context-package-input",
"permission_filtered_count": retrieval.metadata.get("permission_filtered_count", 0),
},
}
return data
def _markitect_context_item(item: MemoryRetrievalItem) -> dict[str, Any]:
node = item.node
source = node.source_spans[0].to_dict() if node.source_spans else {
"path": f"memory://{node.graph_id}/nodes/{node.node_id}",
"unit_kind": node.kind,
"selector": f"nodes[id={node.contract_node_id}]",
"engine": "kontextual-memory",
}
return {
"source": source,
"text": node.text,
"summary": node.metadata.get("summary") or node.metadata.get("title"),
"policy": dict(node.policy),
"provenance": [
{
"kind": "kontextual-memory-export",
"graph_id": node.graph_id,
"node_id": node.node_id,
"contract_node_id": node.contract_node_id,
},
*[dict(entry) for entry in node.provenance],
],
"metadata": {
"memory_runtime": {
"graph_id": node.graph_id,
"contract_graph_id": node.contract_graph_id,
"node_id": node.node_id,
"contract_node_id": node.contract_node_id,
"kind": node.kind,
"edges": [edge.to_dict() for edge in item.edges],
},
**dict(node.metadata),
},
}
def _optional_str(value: Any) -> str | None:
if value is None:
return None

86
tests/test_memory_graph_runtime.py
View File

@@ -8,10 +8,13 @@ from kontextual_engine import (
LifecycleState,
MemoryCompactionRequest,
MemoryGraphImportResult,
MemoryNodeUpdateInstruction,
MemoryPackageExportRequest,
MemoryRefreshRequest,
MemoryRetentionRequest,
MemoryQueryRequest,
MemoryRuntimeService,
MemoryUpdateRequest,
OperationContext,
PolicyDecision,
ValidationError,
@@ -197,6 +200,89 @@ def test_memory_compaction_creates_summary_preserves_spans_and_retires_sources()
assert {node.lifecycle for node in source_nodes} == {LifecycleState.RETIRED}
def test_agent_safe_memory_update_plans_dry_run_and_requires_review_before_write() -> None:
repo = InMemoryMemoryGraphRepository()
service = MemoryRuntimeService(repo)
summary = service.import_markitect_graph(_graph_contract())
context = operation_context()
plan = service.plan_memory_update(
MemoryUpdateRequest(
graph_id=summary.graph_id,
instructions=(
MemoryNodeUpdateInstruction(
contract_node_id="claim.agent-safe-update",
kind="claim",
text="Agent-safe update is planned before durable write.",
source_spans=(
{
"path": "docs/memory-agent-safe.md",
"unit_kind": "section",
"selector": "sections[heading=Agent safe]",
"engine": "selector",
},
),
policy={"labels": ["public"]},
),
),
require_review=True,
reason="capture implementation decision",
),
context,
)
pending = service.apply_memory_update(plan, context)
assert plan.success is True
assert plan.dry_run is True
assert plan.review_required is True
assert plan.planned_updates[0].action == "create_node"
assert plan.planned_updates[0].source_explanation[0]["path"] == "docs/memory-agent-safe.md"
assert pending.success is False
assert pending.review_required is True
assert repo.list_memory_nodes(graph_id=summary.graph_id, kind="claim") == []
approved = service.apply_memory_update(plan, context, review_decision="approved")
assert approved.success is True
assert approved.appended_events[0].kind == "updated"
assert repo.list_memory_nodes(graph_id=summary.graph_id, kind="claim")[0].contract_node_id == (
"claim.agent-safe-update"
)
def test_memory_package_export_emits_markitect_context_package_inputs_without_denied_content() -> None:
repo = InMemoryMemoryGraphRepository()
service = MemoryRuntimeService(repo, policy_gateway=DenyInternalMemoryPolicy())
graph = _graph_contract()
graph["nodes"].append(
{
"id": "claim.internal-secret",
"kind": "claim",
"text": "secret memory text must not leak",
"policy": {"labels": ["internal"]},
}
)
summary = service.import_markitect_graph(graph)
export = service.export_context_package_inputs(
MemoryPackageExportRequest(
query=MemoryQueryRequest(graph_id=summary.graph_id, text_contains="memory"),
title="Memory Package Input",
intent="Export allowed runtime memories for Markitect packaging.",
namespace={"project": "kontextual-engine"},
budget={"max_items": 3},
),
operation_context(),
)
assert export.success is True
assert export.package_input["schema_version"] == "markitect.context-package.input.v1"
assert export.package_input["retrieval_recipes"][0]["kind"] == "memory-runtime-query"
assert export.package_input["items"][0]["source"]["path"] == "workplans/MKTT-WP-0016.md"
assert export.package_input["metadata"]["permission_filtered_count"] == 1
assert "secret memory text must not leak" not in str(export.to_dict())
assert export.audit_event is not None
assert export.audit_event.operation == "memory.package.export"
def test_memory_runtime_service_rejects_invalid_edge_contracts() -> None:
repo = InMemoryMemoryGraphRepository()
service = MemoryRuntimeService(repo)

7
workplans/KONT-WP-0017-agentic-memory-graph-runtime.md
View File

@@ -93,6 +93,11 @@ to `delete_requested`; `refresh_memory()` clears stale markers; and
optionally retiring the source nodes. Each mutating operation appends a memory
event and returns audit metadata.
The agent-safe API slice is implemented with dry-run update plans, explicit
review gates for durable writes, source/policy explanations on planned node
updates, and Markitect-compatible context package input export. The export
method emits package inputs only; Markitect remains the package compiler.
## P17.1 - Import and map Markitect memory contracts
```task
@@ -188,7 +193,7 @@ and tests.
```task
id: KONT-WP-0017-T005
status: todo
status: done
priority: medium
state_hub_task_id: "9e2ad830-c9e6-47a2-8eca-ffe895781fdf"
```