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

memory graph and services

Browse Source
This commit is contained in:
Bernd Worsch
2026-05-15 01:35:58 +02:00
parent 8f732fc868
commit 9d31dcf2af
11 changed files with 1665 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
docs/examples-index.md
View File

@@ -46,6 +46,7 @@ This index maps example files to practical usecases and useful commands.
| `examples/policy/local-label-policy.yaml` | Local policy gateway | `mkt policy check public-agent read note --policy examples/policy/local-label-policy.yaml --label public` |
| `examples/policy/enterprise-policy-map.yaml` | Enterprise IAM mapping fixture | `mkt policy subject examples/policy/netkingdom-claims.yaml --policy-map examples/policy/enterprise-policy-map.yaml` |
| `examples/memory/workplan-context.manifest.yaml` | Context package manifest | `mkt context pack examples/memory/workplan-context.manifest.yaml --root . --no-save` |
| `examples/memory/memory-profile.local.yaml`, `examples/memory/decision-graph*.yaml` | Memory graph/profile contract fixtures | `mkt memory graph pack examples/memory/decision-graph-selection.yaml --format yaml` |
## Literate And Migration

76
docs/memory-graph-contract.md Normal file
View File

@@ -0,0 +1,76 @@
# Memory Graph Contract
Markitect owns the deterministic contract layer for agentic memory. It validates
memory profiles, validates graph snapshots/events, and compiles selected graph
nodes into context packages. Runtime storage, retrieval, compaction, refresh, and
benchmark execution stay outside this repository.
## Contract Versions
- `markitect.memory.profile.v1`: service-agnostic memory blueprint/profile.
- `markitect.memory.graph.v1`: graph snapshot with typed nodes, edges, and event
envelopes.
- `markitect.memory.selection.v1`: packageable graph selection.
## Profile Shape
A profile declares which memory kinds a runtime supports and the contracts that
go with them.
```yaml
schema_version: markitect.memory.profile.v1
id: local-agent-memory
title: Local Agent Memory
intent: Compile selected reasoning and knowledge memories into context packages.
memory_kinds: [reasoning, knowledge, package]
stores:
reasoning: local-event-log
knowledge: local-graph-store
package: markitect-context-registry
activation:
max_items: 8
max_tokens: 2400
```
The profile is descriptive in Markitect. `mkt memory blueprint plan` explains the
handoff to runtime repositories but does not launch services.
## Graph Shape
A graph contains explicit node, edge, and event vocabulary. Unknown kinds fail
validation so that cross-repository contracts do not drift silently.
Valid node kinds include `question`, `claim`, `assumption`, `evidence`,
`decision`, `alternative`, `outcome`, `risk`, `follow_up`, `turn`, `plan`,
`tool_call`, `observation`, `edit`, `validation`, `task`, `topic`, `document`,
`entity`, `artifact`, `concept`, `capability`, `contract`, `policy`,
`preference`, `source_fact`, `episode`, `finding`, `constraint`, `profile`,
`context_package`, and `memory`.
Valid edge kinds include `supports`, `contradicts`, `depends_on`,
`derived_from`, `led_to`, `affects`, `references`, `relates_to`, `supersedes`,
`belongs_to`, `mentions`, `activates`, and `governs`.
Valid event kinds include `recorded`, `updated`, `activated`, `deactivated`,
`refreshed`, `compacted`, `forgotten`, `branched`, `merged`, and
`policy_decision`.
## Selection To Context Package
`mkt memory graph pack` reads a `markitect.memory.selection.v1` file, resolves
the referenced graph/profile, validates both contracts, and emits a normal
`ContextPackage`. Selected nodes become package items. Selected or implied edges
are preserved in package metadata, while selected events are included as
synthetic package items.
```bash
mkt memory blueprint validate examples/memory/memory-profile.local.yaml
mkt memory blueprint plan examples/memory/memory-profile.local.yaml
mkt memory graph validate examples/memory/decision-graph.yaml
mkt memory graph pack examples/memory/decision-graph-selection.yaml --format yaml
```
This keeps Markitect as the compiler/contract boundary. `kontextual-engine`
should implement runtime stores and event production against these schemas, and
`infospace-bench` should benchmark generated context packages and runtime
retrieval behavior against the same fixtures.

19
examples/memory/decision-graph-selection.yaml Normal file
View File

@@ -0,0 +1,19 @@
schema_version: markitect.memory.selection.v1
graph: decision-graph.yaml
profile: memory-profile.local.yaml
title: Memory Contract Boundary Package
intent: Activate the core contract/runtime boundary decisions for implementation work.
namespace:
project: markitect-tool
task: MKTT-WP-0016
node_ids:
- decision.contract-boundary
- constraint.no-runtime-services
- finding.bench-handoff
event_ids:
- event.initial-contract
budget:
max_items: 4
max_tokens: 1200
metadata:
purpose: implementation-context

46
examples/memory/decision-graph.yaml Normal file
View File

@@ -0,0 +1,46 @@
schema_version: markitect.memory.graph.v1
id: markitect-memory-decisions
title: Markitect Memory Decisions
intent: Preserve the decisions that keep memory graphs inside the contract layer.
namespace:
project: markitect-tool
task: MKTT-WP-0016
nodes:
- id: decision.contract-boundary
kind: decision
text: Markitect validates memory profiles and graph snapshots, then compiles selected graph nodes into context packages.
source_spans:
- path: workplans/MKTT-WP-0016-agentic-memory-graphs-and-service-blueprints.md
unit_kind: section
selector: tasks[id=P16.5]
engine: selector
metadata:
title: Contract boundary decision
summary: Markitect remains the deterministic contract and compiler layer.
- id: constraint.no-runtime-services
kind: constraint
text: Runtime storage, retrieval, refresh, and compaction belong in kontextual-engine, not markitect-tool.
metadata:
title: Runtime ownership constraint
- id: finding.bench-handoff
kind: finding
text: infospace-bench should measure package quality, retrieval latency, budget pressure, and regressions using the same graph/profile fixtures.
metadata:
title: Benchmark handoff
edges:
- id: edge.boundary-to-runtime
kind: supports
source: constraint.no-runtime-services
target: decision.contract-boundary
- id: edge.boundary-to-bench
kind: references
source: decision.contract-boundary
target: finding.bench-handoff
events:
- id: event.initial-contract
kind: recorded
timestamp: "2026-05-15T00:00:00Z"
task: MKTT-WP-0016
node_updates:
- node_id: decision.contract-boundary
operation: create

45
examples/memory/memory-profile.local.yaml Normal file
View File

@@ -0,0 +1,45 @@
schema_version: markitect.memory.profile.v1
id: local-agent-memory
title: Local Agent Memory
intent: Validate local memory contracts and compile selected graph context.
memory_kinds:
- reasoning
- knowledge
- package
stores:
reasoning: local-event-log
knowledge: local-graph-store
package: markitect-context-registry
limits:
reasoning:
max_nodes: 200
knowledge:
max_nodes: 1000
package:
max_items: 8
latency:
reasoning:
target_ms: 50
knowledge:
target_ms: 120
retention:
reasoning:
strategy: summarize-after-window
window_events: 50
knowledge:
strategy: explicit-forget-or-supersede
refresh:
cadence: manual
compaction:
strategy: event-summary-layer
activation:
max_items: 6
max_tokens: 1800
reserve_tokens: 200
policy:
required_labels:
- public
observability:
emit_events: true
failure:
missing_store: degrade-to-context-package

228
src/markitect_tool/cli/main.py
View File

@@ -70,11 +70,18 @@ from markitect_tool.memory import (
LocalContextPackageRegistry,
MemoryNamespace,
activate_context_package,
compile_memory_graph_selection_to_context_package,
create_context_package_from_index,
create_context_package_from_manifest,
create_context_package_from_sources,
explain_context_package,
load_memory_graph_file,
load_memory_graph_selection_file,
load_memory_profile_file,
plan_memory_profile,
refresh_context_package,
validate_memory_graph,
validate_memory_profile,
)
from markitect_tool.ops import IncludeError, compose_files, resolve_includes, transform_markdown
from markitect_tool.processor import ProcessorContext, run_fenced_processors
@@ -2012,6 +2019,166 @@ def context_list(root: Path, output_format: str) -> None:
_emit_context_package_list({"count": len(packages), "packages": packages}, output_format)
@main.group("memory")
def memory_group() -> None:
"""Validate memory contracts and compile graph context packages."""
@memory_group.group("blueprint")
def memory_blueprint_group() -> None:
"""Validate and plan memory blueprint/profile contracts."""
@memory_blueprint_group.command("validate")
@click.argument("profile_file", type=click.Path(exists=True, dir_okay=False, path_type=Path))
@click.option(
"--format",
"output_format",
type=click.Choice(["json", "yaml", "text"], case_sensitive=False),
default="text",
show_default=True,
)
def memory_blueprint_validate(profile_file: Path, output_format: str) -> None:
"""Validate a memory profile/blueprint contract."""
try:
profile = load_memory_profile_file(profile_file)
result = validate_memory_profile(profile, path=profile_file)
except ContextPackageError as exc:
raise click.ClickException(str(exc)) from exc
_emit_memory_validation(result.to_dict(), output_format)
if not result.valid:
raise click.ClickException("Memory blueprint validation failed.")
@memory_blueprint_group.command("plan")
@click.argument("profile_file", type=click.Path(exists=True, dir_okay=False, path_type=Path))
@click.option(
"--format",
"output_format",
type=click.Choice(["json", "yaml", "text"], case_sensitive=False),
default="text",
show_default=True,
)
def memory_blueprint_plan(profile_file: Path, output_format: str) -> None:
"""Explain runtime responsibilities implied by a memory profile."""
try:
profile = load_memory_profile_file(profile_file)
result = validate_memory_profile(profile, path=profile_file)
if not result.valid:
_emit_memory_validation(result.to_dict(), output_format)
raise click.ClickException("Memory blueprint validation failed.")
plan = plan_memory_profile(profile)
except ContextPackageError as exc:
raise click.ClickException(str(exc)) from exc
_emit_memory_profile_plan(plan, output_format)
@memory_group.group("graph")
def memory_graph_group() -> None:
"""Validate memory graphs and compile graph selections."""
@memory_graph_group.command("validate")
@click.argument("graph_file", type=click.Path(exists=True, dir_okay=False, path_type=Path))
@click.option(
"--format",
"output_format",
type=click.Choice(["json", "yaml", "text"], case_sensitive=False),
default="text",
show_default=True,
)
def memory_graph_validate(graph_file: Path, output_format: str) -> None:
"""Validate a memory graph contract."""
try:
graph = load_memory_graph_file(graph_file)
result = validate_memory_graph(graph, path=graph_file)
except ContextPackageError as exc:
raise click.ClickException(str(exc)) from exc
_emit_memory_validation(result.to_dict(), output_format)
if not result.valid:
raise click.ClickException("Memory graph validation failed.")
@memory_graph_group.command("pack")
@click.argument("selection_file", type=click.Path(exists=True, dir_okay=False, path_type=Path))
@click.option(
"--graph",
"graph_file",
type=click.Path(exists=True, dir_okay=False, path_type=Path),
help="Override the graph path declared in the selection file.",
)
@click.option(
"--profile",
"profile_file",
type=click.Path(exists=True, dir_okay=False, path_type=Path),
help="Override the profile path declared in the selection file.",
)
@click.option(
"--root",
type=click.Path(exists=True, file_okay=False, path_type=Path),
default=Path("."),
show_default=True,
help="Root used for the local context registry when saving.",
)
@click.option("--output", type=click.Path(dir_okay=False, path_type=Path), help="Write package YAML to this file.")
@click.option("--save", is_flag=True, help="Save the compiled package to the local context registry.")
@click.option(
"--format",
"output_format",
type=click.Choice(["json", "yaml", "text"], case_sensitive=False),
default="text",
show_default=True,
)
def memory_graph_pack(
selection_file: Path,
graph_file: Path | None,
profile_file: Path | None,
root: Path,
output: Path | None,
save: bool,
output_format: str,
) -> None:
"""Compile a memory graph selection into a context package."""
try:
selection = load_memory_graph_selection_file(selection_file)
graph_path = graph_file or _resolve_memory_contract_path(selection_file, selection.graph, "graph")
profile_path = profile_file
if profile_path is None and selection.profile:
profile_path = _resolve_memory_contract_path(selection_file, selection.profile, "profile")
graph = load_memory_graph_file(graph_path)
profile = load_memory_profile_file(profile_path) if profile_path else None
graph_result = validate_memory_graph(graph, path=graph_path)
if not graph_result.valid:
_emit_memory_validation(graph_result.to_dict(), output_format)
raise click.ClickException("Memory graph validation failed.")
if profile:
profile_result = validate_memory_profile(profile, path=profile_path)
if not profile_result.valid:
_emit_memory_validation(profile_result.to_dict(), output_format)
raise click.ClickException("Memory blueprint validation failed.")
package = compile_memory_graph_selection_to_context_package(graph, selection, profile=profile)
except ContextPackageError as exc:
raise click.ClickException(str(exc)) from exc
registry_path = None
if save:
registry_path = LocalContextPackageRegistry(root).save(package)
if output:
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(yaml.safe_dump(package.to_dict(), sort_keys=False), encoding="utf-8")
_emit_context_package(
package.to_dict()
| {
"registry_path": str(registry_path) if registry_path else None,
"output_path": str(output) if output else None,
},
output_format,
)
@main.group()
def workflow() -> None:
"""Inspect, plan, and run declarative Markdown workflows."""
@@ -2800,6 +2967,67 @@ def _emit_context_package_list(data: dict, output_format: str) -> None:
click.echo(f"- {package['id']} {package.get('title', '')}")
def _emit_memory_validation(data: dict, output_format: str) -> None:
if output_format == "json":
click.echo(json.dumps(data, indent=2, ensure_ascii=False))
elif output_format == "yaml":
click.echo(yaml.safe_dump(data, sort_keys=False))
else:
click.echo("valid" if data.get("valid") else "invalid")
click.echo(f"subject: {data.get('subject_kind')} {data.get('subject_id') or ''}".rstrip())
metadata = data.get("metadata", {})
if metadata.get("nodes") is not None:
click.echo(
"graph: "
f"nodes={metadata.get('nodes', 0)} "
f"edges={metadata.get('edges', 0)} "
f"events={metadata.get('events', 0)}"
)
if metadata.get("memory_kinds"):
click.echo("memory_kinds: " + ", ".join(metadata["memory_kinds"]))
for diagnostic in data.get("diagnostics", []):
click.echo(f"! [{diagnostic['severity']}] {diagnostic['code']}: {diagnostic['message']}")
def _emit_memory_profile_plan(data: dict, output_format: str) -> None:
if output_format == "json":
click.echo(json.dumps(data, indent=2, ensure_ascii=False))
elif output_format == "yaml":
click.echo(yaml.safe_dump(data, sort_keys=False))
else:
profile = data.get("profile", {})
click.echo(f"profile: {profile.get('id', '')}")
if profile.get("title"):
click.echo(f"title: {profile['title']}")
click.echo("services_launched_by_markitect_tool: false")
if data.get("activation"):
activation = data["activation"]
click.echo(
"activation_budget: "
f"max_items={activation.get('max_items')} "
f"max_tokens={activation.get('max_tokens')}"
)
click.echo("memory_kinds:")
for plan in data.get("memory_kinds", []):
store = f" store={plan.get('store')}" if plan.get("store") else ""
click.echo(f"- {plan.get('kind')}{store}")
runtime = data.get("runtime_boundary", {})
if runtime:
click.echo("runtime_boundary:")
for owner in ("markitect_tool", "kontextual_engine", "infospace_bench"):
if runtime.get(owner):
click.echo(f"- {owner}: " + "; ".join(runtime[owner]))
def _resolve_memory_contract_path(base_file: Path, raw_path: str | None, label: str) -> Path:
if not raw_path:
raise ContextPackageError(f"Memory graph selection must declare a {label} path or use --{label}.")
candidate = Path(raw_path)
if candidate.is_absolute():
return candidate
return base_file.parent / candidate
def _emit_policy_summary(policy_data: dict) -> None:
click.echo(
"policy: "

58
src/markitect_tool/extension/builtins.py
View File

@@ -25,6 +25,7 @@ def builtin_extension_registry() -> ExtensionRegistry:
_runtime_assessment_descriptor(),
_local_label_policy_descriptor(),
_document_function_descriptor(),
_memory_graph_contract_descriptor(),
_agent_memory_descriptor(),
source_adapter_registry_descriptor(),
]:
@@ -368,3 +369,60 @@ def _agent_memory_descriptor() -> ExtensionDescriptor:
"default_registry": ".markitect/context",
},
)
def _memory_graph_contract_descriptor() -> ExtensionDescriptor:
return ExtensionDescriptor(
id="memory.graph-contract",
kind="memory-contract",
summary="Service-agnostic memory graph/profile contracts and graph-to-context-package compilation.",
capabilities=[
ProcessingCapability(id="memory_profiles", kind="validate"),
ProcessingCapability(id="memory_profiles", kind="plan"),
ProcessingCapability(id="memory_graphs", kind="validate"),
ProcessingCapability(id="context_packages", kind="compile"),
ProcessingCapability(id="provenance", kind="emit"),
ProcessingCapability(id="diagnostics", kind="emit"),
],
safety={
"reads_files": True,
"writes_output_files": True,
"writes_local_context_registry": "optional",
"network": False,
"assisted_generation": False,
"external_memory_services": False,
},
input_contract="MemoryProfile | MemoryGraph | MemoryGraphSelection",
output_contract="MemoryValidationResult | MemoryProfilePlan | ContextPackage",
diagnostics_namespace="memory.graph",
provenance_prefix="memory.graph_contract",
cli={
"commands": [
"mkt memory blueprint validate",
"mkt memory blueprint plan",
"mkt memory graph validate",
"mkt memory graph pack",
]
},
docs=[
"docs/agentic-memory-graph-blueprint-assessment.md",
"docs/memory-graph-contract.md",
],
examples=[
"examples/memory/memory-profile.local.yaml",
"examples/memory/decision-graph.yaml",
"examples/memory/decision-graph-selection.yaml",
],
metadata={
"schema_versions": [
"markitect.memory.profile.v1",
"markitect.memory.graph.v1",
"markitect.memory.selection.v1",
],
"runtime_execution_required": False,
"runtime_handoff_repositories": [
"kontextual-engine",
"infospace-bench",
],
},
)

38
src/markitect_tool/memory/__init__.py
View File

@@ -1,4 +1,4 @@
"""Agent working-memory context packages."""
"""Agent working-memory context packages and graph contracts."""
from markitect_tool.memory.engine import (
ContextActivation,
@@ -20,6 +20,25 @@ from markitect_tool.memory.engine import (
load_context_package_file,
refresh_context_package,
)
from markitect_tool.memory.graph import (
MEMORY_GRAPH_SCHEMA_VERSION,
MEMORY_PROFILE_SCHEMA_VERSION,
MEMORY_SELECTION_SCHEMA_VERSION,
MemoryEvent,
MemoryGraph,
MemoryGraphEdge,
MemoryGraphNode,
MemoryGraphSelection,
MemoryProfile,
MemoryValidationResult,
compile_memory_graph_selection_to_context_package,
load_memory_graph_file,
load_memory_graph_selection_file,
load_memory_profile_file,
plan_memory_profile,
validate_memory_graph,
validate_memory_profile,
)
__all__ = [
"ContextActivation",
@@ -32,12 +51,29 @@ __all__ = [
"RetrievalRecipe",
"SourceSpan",
"SummaryLayer",
"MEMORY_GRAPH_SCHEMA_VERSION",
"MEMORY_PROFILE_SCHEMA_VERSION",
"MEMORY_SELECTION_SCHEMA_VERSION",
"MemoryEvent",
"MemoryGraph",
"MemoryGraphEdge",
"MemoryGraphNode",
"MemoryGraphSelection",
"activate_context_package",
"MemoryProfile",
"MemoryValidationResult",
"compile_memory_graph_selection_to_context_package",
"create_context_package_from_index",
"create_context_package_from_manifest",
"create_context_package_from_sources",
"deactivate_context_package",
"explain_context_package",
"load_context_package_file",
"load_memory_graph_file",
"load_memory_graph_selection_file",
"load_memory_profile_file",
"plan_memory_profile",
"refresh_context_package",
"validate_memory_graph",
"validate_memory_profile",
]

962
src/markitect_tool/memory/graph.py Normal file
View File

@@ -0,0 +1,962 @@
"""Memory graph/profile contracts and context package compilation."""
from __future__ import annotations
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
from typing import Any
import yaml
from markitect_tool.diagnostics import Diagnostic, SourceLocation, has_error
from markitect_tool.memory.engine import (
ContextBudget,
ContextPackage,
ContextPackageError,
ContextPackageItem,
MemoryNamespace,
RetrievalRecipe,
SourceSpan,
)
MEMORY_GRAPH_SCHEMA_VERSION = "markitect.memory.graph.v1"
MEMORY_PROFILE_SCHEMA_VERSION = "markitect.memory.profile.v1"
MEMORY_SELECTION_SCHEMA_VERSION = "markitect.memory.selection.v1"
MEMORY_NODE_KINDS = {
"assumption",
"alternative",
"artifact",
"capability",
"claim",
"concept",
"constraint",
"contract",
"context_package",
"decision",
"document",
"edit",
"entity",
"episode",
"evidence",
"finding",
"follow_up",
"memory",
"observation",
"outcome",
"plan",
"policy",
"preference",
"profile",
"question",
"risk",
"source_fact",
"task",
"topic",
"tool_call",
"turn",
"validation",
}
MEMORY_EDGE_KINDS = {
"activates",
"affects",
"belongs_to",
"contradicts",
"depends_on",
"derived_from",
"governs",
"led_to",
"mentions",
"references",
"relates_to",
"supersedes",
"supports",
}
MEMORY_EVENT_KINDS = {
"activated",
"branched",
"compacted",
"deactivated",
"forgotten",
"merged",
"policy_decision",
"recorded",
"refreshed",
"updated",
}
MEMORY_PROFILE_KINDS = {
"conversation",
"identity",
"knowledge",
"package",
"reasoning",
"source",
"task",
"tool",
}
@dataclass(frozen=True)
class MemoryValidationResult:
"""Validation outcome for a memory graph/profile contract."""
subject_kind: str
subject_id: str | None
valid: bool
diagnostics: list[Diagnostic] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
return _drop_empty(
{
"subject_kind": self.subject_kind,
"subject_id": self.subject_id,
"valid": self.valid,
"diagnostics": [diagnostic.to_dict() for diagnostic in self.diagnostics],
"metadata": self.metadata,
}
)
@dataclass(frozen=True)
class MemoryGraphNode:
"""A memory graph node that can be compiled into context."""
id: str
kind: str
text: str = ""
namespace: MemoryNamespace = field(default_factory=MemoryNamespace)
source_spans: list[SourceSpan] = field(default_factory=list)
provenance: list[dict[str, Any]] = field(default_factory=list)
freshness: dict[str, Any] = field(default_factory=dict)
confidence: float | None = None
policy: dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
@classmethod
def from_mapping(cls, data: dict[str, Any]) -> "MemoryGraphNode":
spans = data.get("source_spans") or data.get("spans") or data.get("source_span") or []
if isinstance(spans, dict):
spans = [spans]
return cls(
id=str(data.get("id") or ""),
kind=str(data.get("kind") or data.get("type") or ""),
text=str(data.get("text") or data.get("content") or data.get("summary") or ""),
namespace=MemoryNamespace.from_mapping(data.get("namespace") or {}),
source_spans=[SourceSpan.from_mapping(span) for span in _mapping_list(spans)],
provenance=_mapping_list(data.get("provenance")),
freshness=dict(data.get("freshness") or {}),
confidence=_optional_float(data.get("confidence")),
policy=dict(data.get("policy") or {}),
metadata=dict(data.get("metadata") or {}),
)
def to_dict(self) -> dict[str, Any]:
return _drop_empty(
{
"id": self.id,
"kind": self.kind,
"text": self.text,
"namespace": self.namespace.to_dict(),
"source_spans": [span.to_dict() for span in self.source_spans],
"provenance": self.provenance,
"freshness": self.freshness,
"confidence": self.confidence,
"policy": self.policy,
"metadata": self.metadata,
}
)
def context_text(self) -> str:
title = self.metadata.get("title") or self.metadata.get("label")
if title and self.text:
return f"{title}\n\n{self.text}"
if self.text:
return self.text
if title:
return str(title)
return f"{self.kind}: {self.id}"
@dataclass(frozen=True)
class MemoryGraphEdge:
"""A directed relationship between memory graph nodes."""
id: str
kind: str
source: str
target: str
provenance: list[dict[str, Any]] = field(default_factory=list)
freshness: dict[str, Any] = field(default_factory=dict)
confidence: float | None = None
policy: dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
@classmethod
def from_mapping(cls, data: dict[str, Any]) -> "MemoryGraphEdge":
kind = str(data.get("kind") or data.get("type") or "")
source = str(data.get("source") or data.get("from") or "")
target = str(data.get("target") or data.get("to") or "")
edge_id = str(data.get("id") or f"edge:{_short_hash([kind, source, target])}")
return cls(
id=edge_id,
kind=kind,
source=source,
target=target,
provenance=_mapping_list(data.get("provenance")),
freshness=dict(data.get("freshness") or {}),
confidence=_optional_float(data.get("confidence")),
policy=dict(data.get("policy") or {}),
metadata=dict(data.get("metadata") or {}),
)
def to_dict(self) -> dict[str, Any]:
return _drop_empty(asdict(self))
@dataclass(frozen=True)
class MemoryEvent:
"""An append-only event envelope for memory runtime handoff."""
id: str
kind: str
timestamp: str
namespace: MemoryNamespace = field(default_factory=MemoryNamespace)
actor: str | None = None
thread: str | None = None
task: str | None = None
node_updates: list[dict[str, Any]] = field(default_factory=list)
edge_updates: list[dict[str, Any]] = field(default_factory=list)
package_refs: list[str] = field(default_factory=list)
activation_refs: list[str] = field(default_factory=list)
branch: dict[str, Any] = field(default_factory=dict)
policy: dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
@classmethod
def from_mapping(cls, data: dict[str, Any]) -> "MemoryEvent":
kind = str(data.get("kind") or data.get("type") or "recorded")
timestamp = str(data.get("timestamp") or data.get("at") or _now())
event_id = str(data.get("id") or f"event:{_short_hash([kind, timestamp, data.get('metadata')])}")
return cls(
id=event_id,
kind=kind,
timestamp=timestamp,
namespace=MemoryNamespace.from_mapping(data.get("namespace") or {}),
actor=_optional_str(data.get("actor")),
thread=_optional_str(data.get("thread")),
task=_optional_str(data.get("task")),
node_updates=_mapping_list(data.get("node_updates") or data.get("nodes")),
edge_updates=_mapping_list(data.get("edge_updates") or data.get("edges")),
package_refs=_string_list(data.get("package_refs") or data.get("packages")),
activation_refs=_string_list(data.get("activation_refs") or data.get("activations")),
branch=dict(data.get("branch") or {}),
policy=dict(data.get("policy") or {}),
metadata=dict(data.get("metadata") or {}),
)
def to_dict(self) -> dict[str, Any]:
return _drop_empty(
{
"id": self.id,
"kind": self.kind,
"timestamp": self.timestamp,
"namespace": self.namespace.to_dict(),
"actor": self.actor,
"thread": self.thread,
"task": self.task,
"node_updates": self.node_updates,
"edge_updates": self.edge_updates,
"package_refs": self.package_refs,
"activation_refs": self.activation_refs,
"branch": self.branch,
"policy": self.policy,
"metadata": self.metadata,
}
)
@dataclass(frozen=True)
class MemoryGraph:
"""Serializable graph contract for agentic memory state."""
id: str
title: str
intent: str
nodes: list[MemoryGraphNode] = field(default_factory=list)
edges: list[MemoryGraphEdge] = field(default_factory=list)
events: list[MemoryEvent] = field(default_factory=list)
namespace: MemoryNamespace = field(default_factory=MemoryNamespace)
schema_version: str = MEMORY_GRAPH_SCHEMA_VERSION
metadata: dict[str, Any] = field(default_factory=dict)
@classmethod
def from_mapping(cls, data: dict[str, Any]) -> "MemoryGraph":
nodes = _mapping_values(data.get("nodes"))
edges = _mapping_values(data.get("edges"))
events = _mapping_values(data.get("events"))
return cls(
id=str(data.get("id") or ""),
title=str(data.get("title") or data.get("name") or ""),
intent=str(data.get("intent") or data.get("description") or ""),
nodes=[MemoryGraphNode.from_mapping(node) for node in nodes],
edges=[MemoryGraphEdge.from_mapping(edge) for edge in edges],
events=[MemoryEvent.from_mapping(event) for event in events],
namespace=MemoryNamespace.from_mapping(data.get("namespace") or {}),
schema_version=str(data.get("schema_version") or data.get("schema") or MEMORY_GRAPH_SCHEMA_VERSION),
metadata=dict(data.get("metadata") or {}),
)
def to_dict(self) -> dict[str, Any]:
return _drop_empty(
{
"schema_version": self.schema_version,
"id": self.id,
"title": self.title,
"intent": self.intent,
"namespace": self.namespace.to_dict(),
"nodes": [node.to_dict() for node in self.nodes],
"edges": [edge.to_dict() for edge in self.edges],
"events": [event.to_dict() for event in self.events],
"metadata": self.metadata,
}
)
def node_index(self) -> dict[str, MemoryGraphNode]:
return {node.id: node for node in self.nodes}
def edge_index(self) -> dict[str, MemoryGraphEdge]:
return {edge.id: edge for edge in self.edges}
def event_index(self) -> dict[str, MemoryEvent]:
return {event.id: event for event in self.events}
@dataclass(frozen=True)
class MemoryProfile:
"""Service-agnostic memory blueprint/profile contract."""
id: str
title: str
intent: str
memory_kinds: list[str] = field(default_factory=list)
stores: dict[str, Any] = field(default_factory=dict)
limits: dict[str, Any] = field(default_factory=dict)
latency: dict[str, Any] = field(default_factory=dict)
retention: dict[str, Any] = field(default_factory=dict)
refresh: dict[str, Any] = field(default_factory=dict)
compaction: dict[str, Any] = field(default_factory=dict)
activation: ContextBudget = field(default_factory=ContextBudget)
policy: dict[str, Any] = field(default_factory=dict)
observability: dict[str, Any] = field(default_factory=dict)
failure: dict[str, Any] = field(default_factory=dict)
schema_version: str = MEMORY_PROFILE_SCHEMA_VERSION
metadata: dict[str, Any] = field(default_factory=dict)
@classmethod
def from_mapping(cls, data: dict[str, Any]) -> "MemoryProfile":
profile_blocks = dict(data.get("profiles") or {})
memory_kinds = _string_list(
data.get("memory_kinds") or data.get("enabled_memory_kinds") or list(profile_blocks.keys())
)
stores = dict(data.get("stores") or {})
limits = dict(data.get("limits") or {})
latency = dict(data.get("latency") or data.get("latency_targets") or {})
retention = dict(data.get("retention") or data.get("retention_policy") or {})
for kind, block in profile_blocks.items():
if not isinstance(block, dict):
continue
if block.get("store") is not None and kind not in stores:
stores[kind] = block["store"]
limit_values = {key: block[key] for key in ("max_nodes", "max_tokens", "max_items") if key in block}
if limit_values and kind not in limits:
limits[kind] = limit_values
if block.get("latency") is not None and kind not in latency:
latency[kind] = block["latency"]
if block.get("retention") is not None and kind not in retention:
retention[kind] = block["retention"]
activation_data = (
data.get("activation")
or data.get("context_budget")
or (data.get("outputs") or {}).get("context_packages")
or {}
)
if isinstance(activation_data, dict) and "budget" in activation_data:
activation_data = activation_data["budget"]
return cls(
id=str(data.get("id") or ""),
title=str(data.get("title") or data.get("name") or ""),
intent=str(data.get("intent") or data.get("description") or ""),
memory_kinds=memory_kinds,
stores=stores,
limits=limits,
latency=latency,
retention=retention,
refresh=dict(data.get("refresh") or {}),
compaction=dict(data.get("compaction") or {}),
activation=ContextBudget.from_mapping(activation_data if isinstance(activation_data, dict) else {}),
policy=dict(data.get("policy") or {}),
observability=dict(data.get("observability") or {}),
failure=dict(data.get("failure") or {}),
schema_version=str(data.get("schema_version") or data.get("schema") or MEMORY_PROFILE_SCHEMA_VERSION),
metadata=dict(data.get("metadata") or {}),
)
def to_dict(self) -> dict[str, Any]:
return _drop_empty(
{
"schema_version": self.schema_version,
"id": self.id,
"title": self.title,
"intent": self.intent,
"memory_kinds": self.memory_kinds,
"stores": self.stores,
"limits": self.limits,
"latency": self.latency,
"retention": self.retention,
"refresh": self.refresh,
"compaction": self.compaction,
"activation": self.activation.to_dict(),
"policy": self.policy,
"observability": self.observability,
"failure": self.failure,
"metadata": self.metadata,
}
)
@dataclass(frozen=True)
class MemoryGraphSelection:
"""A packageable selection of a memory graph."""
graph: str | None = None
profile: str | None = None
title: str | None = None
intent: str | None = None
package_id: str | None = None
namespace: MemoryNamespace = field(default_factory=MemoryNamespace)
node_ids: list[str] = field(default_factory=list)
edge_ids: list[str] = field(default_factory=list)
event_ids: list[str] = field(default_factory=list)
budget: ContextBudget = field(default_factory=ContextBudget)
metadata: dict[str, Any] = field(default_factory=dict)
schema_version: str = MEMORY_SELECTION_SCHEMA_VERSION
@classmethod
def from_mapping(cls, data: dict[str, Any]) -> "MemoryGraphSelection":
return cls(
graph=_optional_str(data.get("graph") or data.get("graph_path") or data.get("source")),
profile=_optional_str(data.get("profile") or data.get("profile_path") or data.get("blueprint")),
title=_optional_str(data.get("title")),
intent=_optional_str(data.get("intent") or data.get("description")),
package_id=_optional_str(data.get("package_id")),
namespace=MemoryNamespace.from_mapping(data.get("namespace") or {}),
node_ids=_string_list(data.get("node_ids") or data.get("nodes")),
edge_ids=_string_list(data.get("edge_ids") or data.get("edges")),
event_ids=_string_list(data.get("event_ids") or data.get("events")),
budget=ContextBudget.from_mapping(data.get("budget") or data.get("context_budget") or {}),
metadata=dict(data.get("metadata") or {}),
schema_version=str(data.get("schema_version") or data.get("schema") or MEMORY_SELECTION_SCHEMA_VERSION),
)
def to_dict(self) -> dict[str, Any]:
return _drop_empty(
{
"schema_version": self.schema_version,
"graph": self.graph,
"profile": self.profile,
"title": self.title,
"intent": self.intent,
"package_id": self.package_id,
"namespace": self.namespace.to_dict(),
"node_ids": self.node_ids,
"edge_ids": self.edge_ids,
"event_ids": self.event_ids,
"budget": self.budget.to_dict(),
"metadata": self.metadata,
}
)
def load_memory_graph_file(path: Path) -> MemoryGraph:
"""Load a memory graph YAML/JSON file."""
return MemoryGraph.from_mapping(_load_mapping_file(path, "Memory graph"))
def load_memory_profile_file(path: Path) -> MemoryProfile:
"""Load a memory profile/blueprint YAML/JSON file."""
return MemoryProfile.from_mapping(_load_mapping_file(path, "Memory profile"))
def load_memory_graph_selection_file(path: Path) -> MemoryGraphSelection:
"""Load a memory graph selection YAML/JSON file."""
return MemoryGraphSelection.from_mapping(_load_mapping_file(path, "Memory graph selection"))
def validate_memory_graph(graph: MemoryGraph | dict[str, Any], path: Path | None = None) -> MemoryValidationResult:
"""Validate a memory graph against the Markitect graph vocabulary."""
loaded = graph if isinstance(graph, MemoryGraph) else MemoryGraph.from_mapping(graph)
diagnostics: list[Diagnostic] = []
contract = SourceLocation(path=str(path)) if path else None
if not loaded.id:
diagnostics.append(_error("memory.graph.missing_id", "Memory graph must declare an id.", contract))
if loaded.schema_version != MEMORY_GRAPH_SCHEMA_VERSION:
diagnostics.append(
_warning(
"memory.graph.schema_version",
f"Expected schema_version `{MEMORY_GRAPH_SCHEMA_VERSION}`.",
contract,
{"actual": loaded.schema_version},
)
)
if not loaded.nodes:
diagnostics.append(_warning("memory.graph.empty", "Memory graph has no nodes.", contract))
node_ids = [node.id for node in loaded.nodes]
_add_duplicate_diagnostics(diagnostics, node_ids, "memory.graph.duplicate_node", "Duplicate node id", contract)
for node in loaded.nodes:
if not node.id:
diagnostics.append(_error("memory.graph.node.missing_id", "Memory graph node must declare an id.", contract))
if node.kind not in MEMORY_NODE_KINDS:
diagnostics.append(
_error(
"memory.graph.node.unknown_kind",
f"Unknown memory node kind `{node.kind}`.",
contract,
{"node_id": node.id, "allowed": sorted(MEMORY_NODE_KINDS)},
)
)
known_nodes = {node.id for node in loaded.nodes if node.id}
edge_ids = [edge.id for edge in loaded.edges]
_add_duplicate_diagnostics(diagnostics, edge_ids, "memory.graph.duplicate_edge", "Duplicate edge id", contract)
for edge in loaded.edges:
if edge.kind not in MEMORY_EDGE_KINDS:
diagnostics.append(
_error(
"memory.graph.edge.unknown_kind",
f"Unknown memory edge kind `{edge.kind}`.",
contract,
{"edge_id": edge.id, "allowed": sorted(MEMORY_EDGE_KINDS)},
)
)
if edge.source not in known_nodes:
diagnostics.append(
_error(
"memory.graph.edge.unknown_source",
f"Edge `{edge.id}` references unknown source node `{edge.source}`.",
contract,
{"edge_id": edge.id},
)
)
if edge.target not in known_nodes:
diagnostics.append(
_error(
"memory.graph.edge.unknown_target",
f"Edge `{edge.id}` references unknown target node `{edge.target}`.",
contract,
{"edge_id": edge.id},
)
)
event_ids = [event.id for event in loaded.events]
_add_duplicate_diagnostics(diagnostics, event_ids, "memory.graph.duplicate_event", "Duplicate event id", contract)
for event in loaded.events:
if event.kind not in MEMORY_EVENT_KINDS:
diagnostics.append(
_error(
"memory.graph.event.unknown_kind",
f"Unknown memory event kind `{event.kind}`.",
contract,
{"event_id": event.id, "allowed": sorted(MEMORY_EVENT_KINDS)},
)
)
return MemoryValidationResult(
subject_kind="memory_graph",
subject_id=loaded.id or None,
valid=not has_error(diagnostics),
diagnostics=diagnostics,
metadata={
"nodes": len(loaded.nodes),
"edges": len(loaded.edges),
"events": len(loaded.events),
},
)
def validate_memory_profile(
profile: MemoryProfile | dict[str, Any], path: Path | None = None
) -> MemoryValidationResult:
"""Validate a memory profile/blueprint contract."""
loaded = profile if isinstance(profile, MemoryProfile) else MemoryProfile.from_mapping(profile)
diagnostics: list[Diagnostic] = []
contract = SourceLocation(path=str(path)) if path else None
if not loaded.id:
diagnostics.append(_error("memory.profile.missing_id", "Memory profile must declare an id.", contract))
if loaded.schema_version != MEMORY_PROFILE_SCHEMA_VERSION:
diagnostics.append(
_warning(
"memory.profile.schema_version",
f"Expected schema_version `{MEMORY_PROFILE_SCHEMA_VERSION}`.",
contract,
{"actual": loaded.schema_version},
)
)
if not loaded.memory_kinds:
diagnostics.append(
_error("memory.profile.no_memory_kinds", "Memory profile must declare at least one memory kind.", contract)
)
for kind in loaded.memory_kinds:
if kind not in MEMORY_PROFILE_KINDS:
diagnostics.append(
_error(
"memory.profile.unknown_kind",
f"Unknown memory profile kind `{kind}`.",
contract,
{"allowed": sorted(MEMORY_PROFILE_KINDS)},
)
)
if kind not in loaded.stores:
diagnostics.append(
_warning(
"memory.profile.store_missing",
f"Memory kind `{kind}` has no store contract.",
contract,
{"memory_kind": kind},
)
)
if not _has_budget_limit(loaded.activation):
diagnostics.append(
_warning(
"memory.profile.activation_budget_missing",
"Profile has no context package activation budget.",
contract,
)
)
return MemoryValidationResult(
subject_kind="memory_profile",
subject_id=loaded.id or None,
valid=not has_error(diagnostics),
diagnostics=diagnostics,
metadata={
"memory_kinds": loaded.memory_kinds,
"stores": sorted(loaded.stores),
},
)
def plan_memory_profile(profile: MemoryProfile) -> dict[str, Any]:
"""Explain how a memory profile maps to runtime responsibilities."""
kind_plans = []
for kind in profile.memory_kinds:
kind_plans.append(
_drop_empty(
{
"kind": kind,
"store": profile.stores.get(kind),
"limits": profile.limits.get(kind),
"latency": profile.latency.get(kind),
"retention": profile.retention.get(kind),
}
)
)
return _drop_empty(
{
"profile": profile.to_dict(),
"memory_kinds": kind_plans,
"activation": profile.activation.to_dict(),
"runtime_boundary": {
"markitect_tool": [
"validate profile and graph contracts",
"compile selected graph nodes into context packages",
"emit deterministic package/provenance metadata",
],
"kontextual_engine": [
"execute stores, retrieval, refresh, compaction, and policy decisions",
"emit memory events that conform to the envelope",
],
"infospace_bench": [
"measure retrieval quality, latency, budget pressure, and regression behavior",
],
"services_launched_by_markitect_tool": False,
},
"handoff_contracts": [
MEMORY_PROFILE_SCHEMA_VERSION,
MEMORY_GRAPH_SCHEMA_VERSION,
MEMORY_SELECTION_SCHEMA_VERSION,
],
}
)
def compile_memory_graph_selection_to_context_package(
graph: MemoryGraph,
selection: MemoryGraphSelection | dict[str, Any] | None = None,
profile: MemoryProfile | None = None,
) -> ContextPackage:
"""Compile a memory graph selection into a deterministic context package."""
selected = selection
if selected is None:
selected = MemoryGraphSelection()
if isinstance(selected, dict):
selected = MemoryGraphSelection.from_mapping(selected)
graph_validation = validate_memory_graph(graph)
if not graph_validation.valid:
messages = "; ".join(diagnostic.message for diagnostic in graph_validation.diagnostics)
raise ContextPackageError(f"Cannot compile invalid memory graph: {messages}")
if profile:
profile_validation = validate_memory_profile(profile)
if not profile_validation.valid:
messages = "; ".join(diagnostic.message for diagnostic in profile_validation.diagnostics)
raise ContextPackageError(f"Cannot compile with invalid memory profile: {messages}")
node_index = graph.node_index()
edge_index = graph.edge_index()
event_index = graph.event_index()
node_ids = selected.node_ids or [node.id for node in graph.nodes]
missing_nodes = [node_id for node_id in node_ids if node_id not in node_index]
if missing_nodes:
raise ContextPackageError(f"Memory graph selection references unknown node ids: {', '.join(missing_nodes)}")
if selected.edge_ids:
edge_ids = selected.edge_ids
else:
selected_node_ids = set(node_ids)
edge_ids = [
edge.id for edge in graph.edges if edge.source in selected_node_ids and edge.target in selected_node_ids
]
missing_edges = [edge_id for edge_id in edge_ids if edge_id not in edge_index]
if missing_edges:
raise ContextPackageError(f"Memory graph selection references unknown edge ids: {', '.join(missing_edges)}")
missing_events = [event_id for event_id in selected.event_ids if event_id not in event_index]
if missing_events:
raise ContextPackageError(f"Memory graph selection references unknown event ids: {', '.join(missing_events)}")
items = [_context_item_for_node(graph, node_index[node_id]) for node_id in node_ids]
items.extend(_context_item_for_event(graph, event_index[event_id]) for event_id in selected.event_ids)
namespace = selected.namespace if selected.namespace.to_dict() else graph.namespace
metadata = {
"memory_graph": {
"schema_version": graph.schema_version,
"graph_id": graph.id,
"selection": selected.to_dict(),
"selected_nodes": node_ids,
"selected_edges": edge_ids,
"selected_events": selected.event_ids,
"edges": [edge_index[edge_id].to_dict() for edge_id in edge_ids],
},
"memory_profile": profile.to_dict() if profile else None,
}
retrieval = RetrievalRecipe(
kind="memory-graph-selection",
query=",".join(node_ids),
engine="memory-graph",
sources=[selected.graph or f"memory://{graph.id}"],
metadata={
"graph_id": graph.id,
"profile_id": profile.id if profile else None,
"selection_schema": selected.schema_version,
},
)
return ContextPackage.create(
title=selected.title or graph.title or "Memory graph package",
intent=selected.intent or graph.intent or "Selected memory graph context.",
namespace=namespace,
items=items,
retrieval_recipes=[retrieval],
budget=selected.budget if _has_budget_limit(selected.budget) else profile.activation if profile else None,
package_id=selected.package_id,
freshness={"compiled_at": _now(), "source": "memory-graph"},
provenance=[
{
"kind": "memory-graph-selection",
"graph_id": graph.id,
"profile_id": profile.id if profile else None,
}
],
metadata=metadata,
)
def _context_item_for_node(graph: MemoryGraph, node: MemoryGraphNode) -> ContextPackageItem:
source = node.source_spans[0] if node.source_spans else _synthetic_span(graph.id, "nodes", node.id, node.kind)
provenance = [
{
"kind": "memory-graph-node",
"graph_id": graph.id,
"node_id": node.id,
"node_kind": node.kind,
}
] + node.provenance
metadata = {
"memory_graph": {
"graph_id": graph.id,
"node_id": node.id,
"node_kind": node.kind,
"freshness": node.freshness,
"confidence": node.confidence,
},
**node.metadata,
}
return ContextPackageItem.create(
source=source,
text=node.context_text(),
summary=_optional_str(node.metadata.get("summary")),
policy=node.policy,
provenance=provenance,
metadata=metadata,
)
def _context_item_for_event(graph: MemoryGraph, event: MemoryEvent) -> ContextPackageItem:
source = _synthetic_span(graph.id, "events", event.id, event.kind)
text = json.dumps(event.to_dict(), indent=2, ensure_ascii=False)
return ContextPackageItem.create(
source=source,
text=text,
summary=f"{event.kind} event {event.id}",
policy=event.policy,
provenance=[
{
"kind": "memory-graph-event",
"graph_id": graph.id,
"event_id": event.id,
"event_kind": event.kind,
}
],
metadata={"memory_graph": {"graph_id": graph.id, "event_id": event.id, "event_kind": event.kind}},
)
def _synthetic_span(graph_id: str, collection: str, item_id: str, item_kind: str) -> SourceSpan:
return SourceSpan(
path=f"memory://{graph_id}/{collection}/{item_id}",
unit_kind=item_kind,
selector=f"{collection}[id={item_id}]",
engine="memory-graph",
)
def _load_mapping_file(path: Path, label: str) -> dict[str, Any]:
try:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
except OSError as exc:
raise ContextPackageError(str(exc)) from exc
if not isinstance(data, dict):
raise ContextPackageError(f"{label} file `{path}` must contain a mapping.")
return data
def _mapping_values(value: Any) -> list[dict[str, Any]]:
if value is None:
return []
if isinstance(value, dict):
items = []
for key, raw in value.items():
if not isinstance(raw, dict):
continue
item = dict(raw)
item.setdefault("id", str(key))
items.append(item)
return items
return _mapping_list(value)
def _mapping_list(value: Any) -> list[dict[str, Any]]:
if value is None:
return []
if isinstance(value, dict):
return [dict(value)]
if isinstance(value, list):
return [dict(item) for item in value if isinstance(item, dict)]
return []
def _string_list(value: Any) -> list[str]:
if value is None:
return []
if isinstance(value, list):
return [str(item) for item in value if item is not None]
if isinstance(value, tuple):
return [str(item) for item in value if item is not None]
return [str(value)]
def _optional_str(value: Any) -> str | None:
if value is None:
return None
text = str(value)
return text if text else None
def _optional_float(value: Any) -> float | None:
if value is None or value == "":
return None
return float(value)
def _short_hash(value: Any) -> str:
return hashlib.sha256(
json.dumps(value, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8")
).hexdigest()[:16]
def _has_budget_limit(budget: ContextBudget) -> bool:
return budget.max_tokens is not None or budget.max_items is not None or budget.reserve_tokens > 0
def _now() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
def _error(
code: str,
message: str,
contract: SourceLocation | None,
details: dict[str, Any] | None = None,
) -> Diagnostic:
return Diagnostic(severity="error", code=code, message=message, contract=contract, details=details or {})
def _warning(
code: str,
message: str,
contract: SourceLocation | None,
details: dict[str, Any] | None = None,
) -> Diagnostic:
return Diagnostic(severity="warning", code=code, message=message, contract=contract, details=details or {})
def _add_duplicate_diagnostics(
diagnostics: list[Diagnostic],
values: list[str],
code: str,
message: str,
contract: SourceLocation | None,
) -> None:
seen: set[str] = set()
duplicates: set[str] = set()
for value in values:
if not value:
continue
if value in seen:
duplicates.add(value)
seen.add(value)
for duplicate in sorted(duplicates):
diagnostics.append(_error(code, f"{message} `{duplicate}`.", contract, {"id": duplicate}))
def _drop_empty(data: dict[str, Any]) -> dict[str, Any]:
return {
key: value
for key, value in data.items()
if value not in (None, [], {}, "")
}

164
tests/test_memory_graph_contracts.py Normal file
View File

@@ -0,0 +1,164 @@
import json
from pathlib import Path
import yaml
from click.testing import CliRunner
from markitect_tool.cli import main
from markitect_tool.memory import (
compile_memory_graph_selection_to_context_package,
load_memory_graph_file,
load_memory_graph_selection_file,
load_memory_profile_file,
plan_memory_profile,
validate_memory_graph,
validate_memory_profile,
)
def test_memory_profile_validation_and_plan(tmp_path: Path):
profile_path = _write_profile(tmp_path)
profile = load_memory_profile_file(profile_path)
result = validate_memory_profile(profile, path=profile_path)
plan = plan_memory_profile(profile)
assert result.valid
assert result.metadata["memory_kinds"] == ["reasoning", "knowledge", "package"]
assert plan["runtime_boundary"]["services_launched_by_markitect_tool"] is False
assert plan["activation"]["max_items"] == 4
def test_memory_graph_validation_and_context_package_compile(tmp_path: Path):
profile_path = _write_profile(tmp_path)
graph_path = _write_graph(tmp_path)
selection_path = _write_selection(tmp_path)
graph = load_memory_graph_file(graph_path)
profile = load_memory_profile_file(profile_path)
selection = load_memory_graph_selection_file(selection_path)
result = validate_memory_graph(graph, path=graph_path)
package = compile_memory_graph_selection_to_context_package(graph, selection, profile=profile)
assert result.valid
assert package.title == "Decision package"
assert len(package.items) == 1
assert package.items[0].source.path == "decision.md"
assert package.metadata["memory_graph"]["selected_edges"] == ["edge.supports"]
assert package.retrieval_recipes[0].kind == "memory-graph-selection"
def test_mkt_memory_blueprint_and_graph_cli(tmp_path: Path):
_write_profile(tmp_path)
_write_graph(tmp_path)
selection_path = _write_selection(tmp_path)
runner = CliRunner()
blueprint = runner.invoke(
main,
["memory", "blueprint", "validate", str(tmp_path / "profile.yaml"), "--format", "json"],
)
plan = runner.invoke(
main,
["memory", "blueprint", "plan", str(tmp_path / "profile.yaml"), "--format", "json"],
)
graph = runner.invoke(
main,
["memory", "graph", "validate", str(tmp_path / "graph.yaml"), "--format", "json"],
)
packed = runner.invoke(main, ["memory", "graph", "pack", str(selection_path), "--format", "json"])
assert blueprint.exit_code == 0, blueprint.output
assert json.loads(blueprint.output)["valid"] is True
assert plan.exit_code == 0, plan.output
assert json.loads(plan.output)["profile"]["id"] == "test-profile"
assert graph.exit_code == 0, graph.output
assert json.loads(graph.output)["metadata"]["nodes"] == 1
assert packed.exit_code == 0, packed.output
package = json.loads(packed.output)
assert package["title"] == "Decision package"
assert package["metadata"]["memory_profile"]["id"] == "test-profile"
def _write_profile(root: Path) -> Path:
path = root / "profile.yaml"
path.write_text(
yaml.safe_dump(
{
"schema_version": "markitect.memory.profile.v1",
"id": "test-profile",
"title": "Test profile",
"intent": "Exercise local memory profile contracts.",
"memory_kinds": ["reasoning", "knowledge", "package"],
"stores": {
"reasoning": "event-log",
"knowledge": "graph-store",
"package": "context-registry",
},
"activation": {"max_items": 4, "max_tokens": 1000},
},
sort_keys=False,
),
encoding="utf-8",
)
return path
def _write_graph(root: Path) -> Path:
path = root / "graph.yaml"
path.write_text(
yaml.safe_dump(
{
"schema_version": "markitect.memory.graph.v1",
"id": "test-graph",
"title": "Test graph",
"intent": "Compile one decision into context.",
"nodes": [
{
"id": "decision.keep-contract-local",
"kind": "decision",
"text": "Keep memory graph contracts in markitect-tool.",
"source_spans": [
{
"path": "decision.md",
"unit_kind": "section",
"selector": "sections[heading=Decision]",
"engine": "selector",
}
],
}
],
"edges": [
{
"id": "edge.supports",
"kind": "supports",
"source": "decision.keep-contract-local",
"target": "decision.keep-contract-local",
}
],
},
sort_keys=False,
),
encoding="utf-8",
)
return path
def _write_selection(root: Path) -> Path:
path = root / "selection.yaml"
path.write_text(
yaml.safe_dump(
{
"schema_version": "markitect.memory.selection.v1",
"graph": "graph.yaml",
"profile": "profile.yaml",
"title": "Decision package",
"intent": "Activate graph decisions.",
"node_ids": ["decision.keep-contract-local"],
"budget": {"max_items": 2},
},
sort_keys=False,
),
encoding="utf-8",
)
return path

37
workplans/MKTT-WP-0016-agentic-memory-graphs-and-service-blueprints.md
View File

@@ -3,7 +3,7 @@ id: MKTT-WP-0016
type: workplan
title: "Memory Graph Profile Contract And Context Package Compiler"
domain: markitect
status: todo
status: in_progress
owner: markitect-tool
topic_slug: markitect
planning_priority: P2
@@ -102,6 +102,27 @@ or external policy services required. The first version should be deterministic
and local-first, with runtime responsibilities handed off to
`kontextual-engine` or future memory-runtime packages.
## Implementation Update - 2026-05-15
Initial implementation is available in `src/markitect_tool/memory/graph.py`.
It adds deterministic memory profile, graph, event, and selection models;
contract validation helpers; profile planning; and graph-selection compilation
into existing `ContextPackage` objects.
The CLI surface from this workplan is implemented:
```text
mkt memory blueprint validate <file>
mkt memory blueprint plan <file>
mkt memory graph validate <file>
mkt memory graph pack <selection-file>
```
Examples and documentation now live in `docs/memory-graph-contract.md` and
`examples/memory/*graph*.yaml`. The remaining refinement is fixture breadth:
add invalid fixtures and richer conversation/knowledge examples before closing
the workplan fully.
## Out Of Scope
- Durable graph/event persistence.
@@ -117,7 +138,7 @@ and local-first, with runtime responsibilities handed off to
```task
id: MKTT-WP-0016-T001
status: todo
status: done
priority: high
state_hub_task_id: "e1523ffb-47e6-4017-88e5-907586297416"
```
@@ -143,7 +164,7 @@ fixture examples. Runtime storage and graph query execution are out of scope.
```task
id: MKTT-WP-0016-T002
status: todo
status: done
priority: high
state_hub_task_id: "a5b871ea-1f5b-4065-bae2-04765da3bc96"
```
@@ -168,7 +189,7 @@ runtime event logs. Persisting event logs belongs outside `markitect-tool`.
```task
id: MKTT-WP-0016-T003
status: todo
status: done
priority: high
state_hub_task_id: "07c94925-c593-4acb-8f2b-b6c61744deff"
```
@@ -195,7 +216,7 @@ retention, compaction, latency, or policy decisions.
```task
id: MKTT-WP-0016-T004
status: todo
status: in_progress
priority: medium
state_hub_task_id: "afcd7ce9-e657-4779-b3b2-0ae3f8e2d66e"
```
@@ -218,7 +239,7 @@ SQLite graph store or runtime query service in this workplan.
```task
id: MKTT-WP-0016-T005
status: todo
status: done
priority: high
state_hub_task_id: "5dfbe77e-f926-4ce8-a8c3-3c10aec0f2f6"
```
@@ -241,7 +262,7 @@ tests.
```task
id: MKTT-WP-0016-T006
status: todo
status: done
priority: medium
state_hub_task_id: "07fb3f0d-aaf7-4bc0-a7a3-0fd4983776bb"
```
@@ -265,7 +286,7 @@ Output: CLI commands, docs, and tests.
```task
id: MKTT-WP-0016-T007
status: todo
status: in_progress
priority: medium
state_hub_task_id: "ff78d449-0738-4f1b-a018-2efed3d8e878"
```