feat(acms): implement Temporal Data Model (Revision-Aware RDF) with 3 storage tiers #615

Merged
hamza.khyari merged 1 commits from feature/m6-temporal-data-model-revision-aware-rdf into master 2026-03-11 01:43:38 +00:00
18 changed files with 4441 additions and 10 deletions
+14
View File
@@ -80,6 +80,20 @@
output summary) and one regression guard for interactive mode. Includes Robot
Framework smoke tests and ASV benchmarks. Tests are intentionally failing until
the bug fix for #522 is applied. (#536)
- Added Temporal Data Model (Revision-Aware RDF) with 3 storage tiers for
the ACMS. Temporal metadata fields (`valid_from`, `valid_until`,
`is_current`, `is_revision_of`) on UKO InformationUnit nodes enable
revision chain tracking: when code changes, old nodes are marked historical
and new revision nodes are created with back-links. Three storage tiers
(hot/warm/cold) filter nodes by temporal scope (current/recent/all) with
configurable retention (`warm_retention_hours` default 24h,
`cold_retention_days` default 90d). Includes `TemporalMetadata`,
`TemporalNode`, `RevisionChain`, `TierQueryResult`, `TierRetentionConfig`
frozen domain models, `TemporalBackend` protocol,
`InMemoryTemporalBackend` stub, `TemporalService` with structlog and DI,
`BackendSet.temporal` typing upgrade from `object | None` to
`TemporalBackend | None`. 99 Behave scenarios, 8 Robot Framework tests,
ASV benchmarks, and reference documentation. (#577)
- Implemented UKO Layer 1 Domain Ontologies (`uko-doc:`, `uko-data:`, `uko-infra:`)
in the OWL/Turtle ontology file (`docs/ontology/uko.ttl`). Added 17 `uko-doc:` classes
(Document, Section, Paragraph, Citation, etc.), 13 `uko-data:` classes (Table, Column,
+202
View File
@@ -0,0 +1,202 @@
"""ASV benchmarks for Temporal Data Model (Revision-Aware RDF).
Measures the performance of:
- TemporalMetadata / TemporalNode / RevisionChain model creation
- InMemoryTemporalBackend revision operations
- Tier-aware temporal queries
- Revision chain traversal
"""
from __future__ import annotations
import importlib
import sys
from datetime import UTC, datetime, timedelta
from pathlib import Path
# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
# Force-reload so ASV picks up the source tree version.
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.domain.models.acms.temporal import ( # noqa: E402
RevisionChain,
TemporalMetadata,
TemporalNode,
TierRetentionConfig,
)
from cleveragents.domain.models.acms.temporal_stubs import ( # noqa: E402
InMemoryTemporalBackend,
)
from cleveragents.domain.models.acms.tiers import ContextTier # noqa: E402
from cleveragents.domain.models.core.project import ( # noqa: E402
TemporalScope,
)
_NOW = datetime.now(tz=UTC)
# ---------------------------------------------------------------------------
# Model creation benchmarks
# ---------------------------------------------------------------------------
class TemporalModelSuite:
"""Benchmark temporal domain model creation overhead."""
timeout = 60
def time_create_temporal_metadata(self) -> None:
TemporalMetadata(valid_from=_NOW)
def time_create_temporal_metadata_full(self) -> None:
TemporalMetadata(
valid_from=_NOW,
valid_until=_NOW + timedelta(days=1),
is_current=False,
is_revision_of="uko-py:class/Foo_v1",
)
def time_create_temporal_node(self) -> None:
TemporalNode(
node_uri="uko-py:class/Auth_v1",
source_resource="RES001",
source_path="src/auth.py",
temporal=TemporalMetadata(valid_from=_NOW),
)
def time_create_revision_chain(self) -> None:
RevisionChain(
current_uri="uko-py:class/Auth_v3",
predecessors=(
"uko-py:class/Auth_v1",
"uko-py:class/Auth_v2",
),
)
def time_create_tier_retention_config(self) -> None:
TierRetentionConfig()
# ---------------------------------------------------------------------------
# Backend operation benchmarks
# ---------------------------------------------------------------------------
class TemporalBackendSuite:
"""Benchmark InMemoryTemporalBackend operations."""
timeout = 60
def setup(self) -> None:
"""Create a backend with 10 revision chains of 5 versions each."""
self.backend = InMemoryTemporalBackend()
self.retention = TierRetentionConfig()
for chain_idx in range(10):
base = f"uko-py:class/Chain{chain_idx}"
prev_uri: str | None = None
for ver in range(1, 6):
uri = f"{base}_v{ver}"
node = TemporalNode(
node_uri=uri,
source_resource=f"RES{chain_idx:03d}",
source_path=f"src/chain{chain_idx}.py",
temporal=TemporalMetadata(
valid_from=_NOW - timedelta(days=10 - ver),
is_current=(ver == 5),
is_revision_of=prev_uri,
valid_until=(
_NOW - timedelta(days=10 - ver - 1) if ver < 5 else None
),
),
)
self.backend.store_node(node)
prev_uri = uri
def time_get_current(self) -> None:
self.backend.get_current("uko-py:class/Chain5")
def time_get_history_all(self) -> None:
self.backend.get_history(
"uko-py:class/Chain3",
TemporalScope.ALL,
)
def time_get_history_current(self) -> None:
self.backend.get_history(
"uko-py:class/Chain3",
TemporalScope.CURRENT,
)
def time_get_revision_chain(self) -> None:
self.backend.get_revision_chain("uko-py:class/Chain3_v5")
def time_query_hot_tier(self) -> None:
self.backend.query_by_tier(
ContextTier.HOT,
TemporalScope.CURRENT,
self.retention,
)
def time_query_cold_tier(self) -> None:
self.backend.query_by_tier(
ContextTier.COLD,
TemporalScope.ALL,
self.retention,
)
def time_create_revision(self) -> None:
"""Benchmark a single revision creation (resets after)."""
# Store a temporary node to revise
tmp = TemporalNode(
node_uri="uko-py:class/Tmp_v1",
source_resource="RES999",
source_path="src/tmp.py",
temporal=TemporalMetadata(
valid_from=_NOW - timedelta(days=1),
is_current=True,
),
)
self.backend.store_node(tmp)
new = TemporalNode(
node_uri="uko-py:class/Tmp_v2",
source_resource="RES999",
source_path="src/tmp.py",
temporal=TemporalMetadata(
valid_from=_NOW,
is_current=True,
is_revision_of="uko-py:class/Tmp_v1",
),
)
self.backend.create_revision(
"uko-py:class/Tmp_v1",
new,
_NOW,
)
# Clean up for next iteration
del self.backend._nodes["uko-py:class/Tmp_v1"]
del self.backend._nodes["uko-py:class/Tmp_v2"]
def time_mark_historical(self) -> None:
"""Benchmark marking a node historical (resets after)."""
tmp = TemporalNode(
node_uri="uko-py:class/MarkTmp_v1",
source_resource="RES999",
source_path="src/tmp.py",
temporal=TemporalMetadata(
valid_from=_NOW,
is_current=True,
),
)
self.backend.store_node(tmp)
self.backend.mark_historical("uko-py:class/MarkTmp_v1", _NOW)
del self.backend._nodes["uko-py:class/MarkTmp_v1"]
+121
View File
@@ -0,0 +1,121 @@
# Temporal Data Model (Revision-Aware RDF)
The Temporal Data Model enables revision tracking of UKO knowledge graph nodes
across three storage tiers (hot/warm/cold). When code changes, existing nodes
are **not deleted** -- they are marked historical and a new revision node is
created with a back-link to the predecessor.
Based on `docs/specification.md` lines 41940--42499.
## Architecture
```
Code Change ──► Analyzer ──► TemporalService.create_revision()
┌──────────┴──────────┐
│ TemporalBackend │
│ ┌─────────────────┐ │
│ │ Old node: │ │
│ │ isCurrent=false│ │
│ │ validUntil=now │ │
│ ├─────────────────┤ │
│ │ New node: │ │
│ │ isCurrent=true │ │
│ │ isRevisionOf= │ │
│ │ old_uri │ │
│ └─────────────────┘ │
└──────────────────────┘
```
## Domain Models
All models are frozen Pydantic `BaseModel` instances (immutable value objects).
### TemporalMetadata
Temporal fields carried by every UKO InformationUnit. Maps to the OWL
temporal properties defined in UKO Layer 0.
| Field | Type | Default | Spec Property |
|-------|------|---------|---------------|
| `valid_from` | `datetime` (UTC) | **required** | `uko:validFrom` |
| `valid_until` | `datetime \| None` | `None` | `uko:validUntil` |
| `is_current` | `bool` | `True` | `uko:isCurrent` |
| `is_revision_of` | `str \| None` | `None` | `uko:isRevisionOf` |
### TemporalNode
A UKO InformationUnit with full temporal metadata and provenance.
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `node_uri` | `str` | **required** | Unique UKO URI (e.g., `uko-py:class/Auth_v2`) |
| `source_resource` | `str` | **required** | ULID of originating resource |
| `source_path` | `str` | **required** | File path within the resource |
| `source_range` | `str \| None` | `None` | Source range (e.g., `"15:1-87:0"`) |
| `temporal` | `TemporalMetadata` | **required** | Embedded temporal metadata |
### RevisionChain
Ordered chain of node versions for a single UKO concept.
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `current_uri` | `str` | **required** | URI of the current head node |
| `predecessors` | `tuple[str, ...]` | `()` | Predecessor URIs (oldest-first) |
Properties: `depth` (total versions), `all_uris` (predecessors + current).
### TierRetentionConfig
Retention policy for warm and cold tiers.
| Field | Type | Default | Config Key |
|-------|------|---------|------------|
| `warm_retention_hours` | `int` (≥1) | `24` | `context.tiers.warm.retention-hours` |
| `cold_retention_days` | `int` (≥1) | `90` | `context.tiers.cold.retention-days` |
### TierQueryResult
Result from a tier-aware temporal query.
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `nodes` | `tuple[TemporalNode, ...]` | `()` | Matching nodes |
| `tier` | `ContextTier` | **required** | HOT / WARM / COLD |
| `temporal_scope` | `TemporalScope` | **required** | CURRENT / RECENT / ALL |
## TemporalBackend Protocol
```python
class TemporalBackend(Protocol):
def store_node(self, node) -> None: ...
def create_revision(self, current_uri, new_node, timestamp) -> TemporalNode: ...
def get_current(self, node_uri_base) -> TemporalNode | None: ...
def get_history(self, node_uri_base, temporal_scope) -> tuple[TemporalNode, ...]: ...
def get_revision_chain(self, node_uri) -> RevisionChain: ...
def query_by_tier(self, tier, temporal_scope, retention) -> TierQueryResult: ...
def mark_historical(self, node_uri, timestamp) -> TemporalNode: ...
```
## Three Storage Tiers
| Tier | Content | Temporal Filter | Retention |
|------|---------|-----------------|-----------|
| **Hot** | Current UKO graph | `isCurrent = true` only | Until removed |
| **Warm** | Recent context | Current + recently-expired | `warm_retention_hours` (24h) |
| **Cold** | All historical | All temporal versions | `cold_retention_days` (90d) |
## Revision Chain Mechanics
1. Old node: `is_current``False`, `valid_until` → timestamp
2. New node: `is_current``True`, `valid_from` → timestamp
3. New node: `is_revision_of` → old node URI
4. Old node's `valid_from` is preserved unchanged
## v1 Limitations
- `InMemoryTemporalBackend` stores all nodes in a `dict`; no persistence.
- No SPARQL query helpers yet (deferred to graph backend integration).
- `temporal-archaeology` strategy stub exists but does not yet use the
temporal backend (wiring deferred to strategy integration milestone).
@@ -35,6 +35,7 @@ from cleveragents.domain.models.acms.stubs import (
InMemoryTextBackend,
InMemoryVectorBackend,
)
from cleveragents.domain.models.acms.temporal_stubs import InMemoryTemporalBackend
# ---------------------------------------------------------------------------
# Helpers
@@ -1007,12 +1008,12 @@ def step_then_stats_is_mapping_proxy(context: Context) -> None:
def step_given_backends_graph_temporal(context: Context) -> None:
context.backend_set = BackendSet(
graph=InMemoryGraphBackend(),
temporal=object(), # Presence indicates availability
temporal=InMemoryTemporalBackend(),
)
@given("a BackendSet with temporal backend only")
def step_given_backends_temporal_only(context: Context) -> None:
context.backend_set = BackendSet(
temporal=object(), # Presence indicates availability
temporal=InMemoryTemporalBackend(),
)
@@ -0,0 +1,499 @@
"""Step definitions for Temporal Data Model — backend steps."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from typing import Any
from behave import given, then, when
from cleveragents.domain.models.acms.temporal import (
TemporalMetadata,
TemporalNode,
TierRetentionConfig,
)
from cleveragents.domain.models.acms.temporal_stubs import InMemoryTemporalBackend
from cleveragents.domain.models.acms.tiers import ContextTier
from cleveragents.domain.models.core.project import TemporalScope
__all__: list[str] = []
# ---- InMemoryTemporalBackend steps ----
@given("an InMemoryTemporalBackend instance")
def step_given_inmemory_temporal_backend(context: Any) -> None:
context.temporal_backend = InMemoryTemporalBackend()
@given('a stored temporal node "{uri}" for resource "{resource}" at "{path}"')
def step_given_stored_temporal_node(
context: Any,
uri: str,
resource: str,
path: str,
) -> None:
now = datetime.now(tz=UTC)
node = TemporalNode(
node_uri=uri,
source_resource=resource,
source_path=path,
temporal=TemporalMetadata(
valid_from=now - timedelta(days=5),
is_current=True,
),
)
context.temporal_backend.store_node(node)
context.last_stored_node = node
@when('I get the current node for base "{base}"')
def step_when_get_current(context: Any, base: str) -> None:
context.current_node = context.temporal_backend.get_current(base)
@then('the current node uri should be "{expected}"')
def step_then_current_node_uri(context: Any, expected: str) -> None:
assert context.current_node is not None
assert context.current_node.node_uri == expected
@then("the current node should be None")
def step_then_current_node_none(context: Any) -> None:
assert context.current_node is None
@given('I create a revision from "{old}" to "{new}"')
@when('I create a revision from "{old}" to "{new}"')
def step_when_create_revision(context: Any, old: str, new: str) -> None:
now = datetime.now(tz=UTC)
new_node = TemporalNode(
node_uri=new,
source_resource="RES001",
source_path="src/auth.py",
temporal=TemporalMetadata(
valid_from=now,
is_current=True,
is_revision_of=old,
),
)
context.new_revision = context.temporal_backend.create_revision(
old,
new_node,
now,
)
@then('the old node "{uri}" should not be current')
def step_then_old_node_not_current(context: Any, uri: str) -> None:
node = context.temporal_backend._nodes[uri]
assert node.temporal.is_current is False
@then('the old node "{uri}" should have valid_until set')
def step_then_old_node_valid_until(context: Any, uri: str) -> None:
node = context.temporal_backend._nodes[uri]
assert node.temporal.valid_until is not None
@then('the new node "{uri}" should be current')
def step_then_new_node_current(context: Any, uri: str) -> None:
node = context.temporal_backend._nodes[uri]
assert node.temporal.is_current is True
@then('the new node "{uri}" should have is_revision_of "{expected}"')
def step_then_new_node_revision_of(
context: Any,
uri: str,
expected: str,
) -> None:
node = context.temporal_backend._nodes[uri]
assert node.temporal.is_revision_of == expected
@then('creating a revision from nonexistent "{uri}" should raise KeyError')
def step_then_revision_nonexistent(context: Any, uri: str) -> None:
now = datetime.now(tz=UTC)
new_node = TemporalNode(
node_uri="uko:new",
source_resource="RES",
source_path="src/a.py",
temporal=TemporalMetadata(valid_from=now, is_current=True),
)
try:
context.temporal_backend.create_revision(uri, new_node, now)
msg = "Expected KeyError"
raise AssertionError(msg)
except KeyError:
pass
@then("creating a revision from blank URI should raise ValueError")
def step_then_revision_blank_uri(context: Any) -> None:
now = datetime.now(tz=UTC)
new_node = TemporalNode(
node_uri="uko:new",
source_resource="RES",
source_path="src/a.py",
temporal=TemporalMetadata(valid_from=now, is_current=True),
)
try:
context.temporal_backend.create_revision("", new_node, now)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValueError:
pass
@then('creating a revision from "{uri}" to itself should raise ValueError')
def step_then_revision_same_uri(context: Any, uri: str) -> None:
now = datetime.now(tz=UTC)
same_uri_node = TemporalNode(
node_uri=uri,
source_resource="RES",
source_path="src/a.py",
temporal=TemporalMetadata(valid_from=now, is_current=True),
)
try:
context.temporal_backend.create_revision(uri, same_uri_node, now)
msg = "Expected ValueError for same-URI revision"
raise AssertionError(msg)
except ValueError:
pass
@then('storing a duplicate node "{uri}" should raise ValueError')
def step_then_store_duplicate_raises(context: Any, uri: str) -> None:
now = datetime.now(tz=UTC)
node = TemporalNode(
node_uri=uri,
source_resource="RES_DUP",
source_path="src/dup.py",
temporal=TemporalMetadata(valid_from=now, is_current=True),
)
try:
context.temporal_backend.store_node(node)
msg = "Expected ValueError for duplicate store_node"
raise AssertionError(msg)
except ValueError:
pass
# ---- get_history steps ----
@when('I get the history for base "{base}" with scope ALL')
def step_when_get_history_all(context: Any, base: str) -> None:
context.history = context.temporal_backend.get_history(
base,
TemporalScope.ALL,
)
@when('I get the history for base "{base}" with scope CURRENT')
def step_when_get_history_current(context: Any, base: str) -> None:
context.history = context.temporal_backend.get_history(
base,
TemporalScope.CURRENT,
)
@then("the history should have {count:d} nodes")
def step_then_history_count(context: Any, count: int) -> None:
assert len(context.history) == count
@then("the history should be ordered newest first")
def step_then_history_ordered(context: Any) -> None:
for i in range(len(context.history) - 1):
assert (
context.history[i].temporal.valid_from
>= context.history[i + 1].temporal.valid_from
)
@then("getting history for blank base URI should raise ValueError")
def step_then_history_blank_uri(context: Any) -> None:
try:
context.temporal_backend.get_history("", TemporalScope.ALL)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValueError:
pass
# ---- revision chain steps ----
@when('I get the revision chain for "{uri}"')
def step_when_get_chain(context: Any, uri: str) -> None:
context.chain = context.temporal_backend.get_revision_chain(uri)
@then("the revision chain should have depth {expected:d}")
def step_then_chain_depth_backend(context: Any, expected: int) -> None:
assert context.chain.depth == expected
@then('the revision chain current should be "{expected}"')
def step_then_chain_current(context: Any, expected: str) -> None:
assert context.chain.current_uri == expected
@then('the revision chain predecessors should include "{uri}"')
def step_then_chain_preds_include(context: Any, uri: str) -> None:
assert uri in context.chain.predecessors
@then("getting revision chain for nonexistent URI should raise KeyError")
def step_then_chain_nonexistent(context: Any) -> None:
try:
context.temporal_backend.get_revision_chain("uko:nonexistent")
msg = "Expected KeyError"
raise AssertionError(msg)
except KeyError:
pass
# ---- query by tier steps ----
@when("I query the HOT tier")
def step_when_query_hot(context: Any) -> None:
retention = TierRetentionConfig()
context.tier_query = context.temporal_backend.query_by_tier(
ContextTier.HOT,
TemporalScope.CURRENT,
retention,
)
@when("I query the COLD tier")
def step_when_query_cold(context: Any) -> None:
retention = TierRetentionConfig()
context.tier_query = context.temporal_backend.query_by_tier(
ContextTier.COLD,
TemporalScope.ALL,
retention,
)
@then("the tier query should return {count:d} node")
def step_then_tier_query_count_singular(context: Any, count: int) -> None:
assert len(context.tier_query.nodes) == count
@then("the tier query should return {count:d} nodes")
def step_then_tier_query_count(context: Any, count: int) -> None:
assert len(context.tier_query.nodes) == count
@then("the tier query node should be current")
def step_then_tier_query_node_current(context: Any) -> None:
for node in context.tier_query.nodes:
assert node.temporal.is_current is True
# ---- mark historical steps ----
@given('I mark "{uri}" as historical')
@when('I mark "{uri}" as historical')
def step_when_mark_historical(context: Any, uri: str) -> None:
now = datetime.now(tz=UTC)
context.marked_node = context.temporal_backend.mark_historical(
uri,
now,
)
@then("the marked node should not be current")
def step_then_marked_not_current(context: Any) -> None:
assert context.marked_node.temporal.is_current is False
@then("the marked node should have valid_until set")
def step_then_marked_valid_until(context: Any) -> None:
assert context.marked_node.temporal.valid_until is not None
@then("marking blank URI as historical should raise ValueError")
def step_then_mark_blank_uri(context: Any) -> None:
try:
context.temporal_backend.mark_historical(
"",
datetime.now(tz=UTC),
)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValueError:
pass
@then("marking nonexistent URI as historical should raise KeyError")
def step_then_mark_nonexistent(context: Any) -> None:
try:
context.temporal_backend.mark_historical(
"uko:nonexistent",
datetime.now(tz=UTC),
)
msg = "Expected KeyError"
raise AssertionError(msg)
except KeyError:
pass
# ---- Edge-case guards (Round 3 bug-hunt fixes) ----
@then(
'marking already-historical "{uri}" should raise ValueError',
)
def step_then_mark_already_historical(context: Any, uri: str) -> None:
try:
context.temporal_backend.mark_historical(uri, datetime.now(tz=UTC))
msg = "Expected ValueError for already-historical node"
raise AssertionError(msg)
except ValueError:
pass
@then(
'creating a revision from historical "{current}" '
'to "{new_uri}" should raise ValueError',
)
def step_then_revise_historical_raises(
context: Any,
current: str,
new_uri: str,
) -> None:
now = datetime.now(tz=UTC)
new_node = TemporalNode(
node_uri=new_uri,
source_resource="RES001",
source_path="src/auth.py",
temporal=TemporalMetadata(
valid_from=now,
is_current=True,
is_revision_of=current,
),
)
try:
context.temporal_backend.create_revision(current, new_node, now)
msg = "Expected ValueError for revising historical node"
raise AssertionError(msg)
except ValueError:
pass
@then('the revision chain current_uri should be "{expected}"')
def step_then_chain_current_uri(context: Any, expected: str) -> None:
assert context.chain.current_uri == expected
@then('the revision chain predecessors should contain "{expected}"')
def step_then_chain_predecessors_contain(context: Any, expected: str) -> None:
assert expected in context.chain.predecessors
@given("a stored temporal node with branching revision chain")
def step_given_branching_chain(context: Any) -> None:
"""Create A → B and A → C (sibling branches) via _nodes."""
now = datetime.now(tz=UTC)
base = "uko-py:class/Branch"
nodes = context.temporal_backend._nodes
for suffix, meta_kw in [
(
"_A",
{
"valid_from": now - timedelta(hours=2),
"valid_until": now - timedelta(hours=1),
"is_current": False,
},
),
(
"_B",
{
"valid_from": now - timedelta(hours=1),
"is_current": True,
"is_revision_of": f"{base}_A",
},
),
("_C", {"valid_from": now, "is_current": True, "is_revision_of": f"{base}_A"}),
]:
uri = f"{base}{suffix}"
nodes[uri] = TemporalNode(
node_uri=uri,
source_resource="RES001",
source_path="src/a.py",
temporal=TemporalMetadata(**meta_kw),
)
@then('the revision chain should not contain "{uri}"')
def step_then_chain_not_contain(context: Any, uri: str) -> None:
assert uri not in context.chain.all_uris, (
f"{uri} should not be in chain {context.chain.all_uris}"
)
# ---- Diff-coverage steps (RECENT scope, cycle guard, WARM tier) ----
@when('I get the history for base "{base}" with scope RECENT')
def step_when_get_history_recent(context: Any, base: str) -> None:
context.history = context.temporal_backend.get_history(
base,
TemporalScope.RECENT,
)
@given("a stored temporal node with cyclic revision chain")
def step_given_cyclic_revision_chain(context: Any) -> None:
"""Create two nodes whose is_revision_of pointers form a cycle."""
now = datetime.now(tz=UTC)
for uri, meta_kw in [
(
"uko-py:class/Cycle_A",
{
"valid_from": now - timedelta(days=2),
"valid_until": now - timedelta(days=1),
"is_current": False,
"is_revision_of": "uko-py:class/Cycle_B",
},
),
(
"uko-py:class/Cycle_B",
{
"valid_from": now,
"is_current": True,
"is_revision_of": "uko-py:class/Cycle_A",
},
),
]:
context.temporal_backend.store_node(
TemporalNode(
node_uri=uri,
source_resource="RES001",
source_path="src/cycle.py",
temporal=TemporalMetadata(**meta_kw),
)
)
@given("a stored temporal node with dangling predecessor")
def step_given_dangling_predecessor(context: Any) -> None:
context.temporal_backend.store_node(
TemporalNode(
node_uri="uko-py:class/Dangling_B",
source_resource="RES001",
source_path="src/dangling.py",
temporal=TemporalMetadata(
valid_from=datetime.now(tz=UTC),
is_current=True,
is_revision_of="uko-py:class/Dangling_A",
),
)
)
@@ -0,0 +1,481 @@
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from types import MappingProxyType
from typing import Any
from behave import given, then, when
from pydantic import ValidationError
from cleveragents.domain.models.acms.strategy import BackendSet, StrategyConfig
from cleveragents.domain.models.acms.temporal import (
RevisionChain,
TemporalBackend,
TemporalMetadata,
TemporalNode,
TierQueryResult,
TierRetentionConfig,
)
from cleveragents.domain.models.acms.tiers import ContextTier
from cleveragents.domain.models.core.project import TemporalScope
__all__: list[str] = []
# ---------------------------------------------------------------------------
# RevisionChain steps
# ---------------------------------------------------------------------------
@given('a RevisionChain with only current_uri "{uri}"')
def step_given_revision_chain(context: Any, uri: str) -> None:
context.revision_chain = RevisionChain(current_uri=uri)
@given('a RevisionChain with current_uri "{uri}" and predecessors "{pred_str}"')
def step_given_revision_chain_with_preds(
context: Any,
uri: str,
pred_str: str,
) -> None:
preds = tuple(p.strip() for p in pred_str.split(","))
context.revision_chain = RevisionChain(
current_uri=uri,
predecessors=preds,
)
@then("the revision chain depth should be {expected:d}")
def step_then_chain_depth(context: Any, expected: int) -> None:
assert context.revision_chain.depth == expected
@then("the revision chain predecessors should be empty")
def step_then_chain_preds_empty(context: Any) -> None:
assert context.revision_chain.predecessors == ()
@then("the revision chain predecessors should have {count:d} entries")
def step_then_chain_preds_count(context: Any, count: int) -> None:
assert len(context.revision_chain.predecessors) == count
@then('the revision chain all_uris should contain "{uri}"')
def step_then_chain_all_uris_contains(context: Any, uri: str) -> None:
assert uri in context.revision_chain.all_uris
@then("the revision chain all_uris should have {count:d} entries")
def step_then_chain_all_uris_count(context: Any, count: int) -> None:
assert len(context.revision_chain.all_uris) == count
@then("modifying the revision chain current_uri should raise an error")
def step_then_chain_frozen(context: Any) -> None:
try:
context.revision_chain.current_uri = "changed" # type: ignore[misc]
msg = "Expected frozen model to reject mutation"
raise AssertionError(msg)
except ValidationError:
pass
@then("creating a RevisionChain with empty current_uri should raise ValueError")
def step_then_chain_empty_uri(context: Any) -> None:
try:
RevisionChain(current_uri="")
msg = "Expected ValueError"
raise AssertionError(msg)
except ValidationError:
pass
@then("creating a RevisionChain with whitespace current_uri should raise ValueError")
def step_then_chain_whitespace_uri(context: Any) -> None:
try:
RevisionChain(current_uri=" ")
msg = "Expected ValueError"
raise AssertionError(msg)
except ValidationError:
pass
@then("creating a RevisionChain with empty predecessor should raise ValueError")
def step_then_chain_empty_pred(context: Any) -> None:
try:
RevisionChain(
current_uri="uko:test",
predecessors=("uko:v1", ""),
)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValidationError:
pass
@then(
"creating a RevisionChain with current_uri in predecessors should raise ValueError",
)
def step_then_chain_current_in_preds(context: Any) -> None:
try:
RevisionChain(
current_uri="uko:v2",
predecessors=("uko:v1", "uko:v2"),
)
msg = "Expected ValueError for current_uri in predecessors"
raise AssertionError(msg)
except ValidationError:
pass
@then(
"creating a RevisionChain with duplicate predecessors should raise ValueError",
)
def step_then_chain_duplicate_preds(context: Any) -> None:
try:
RevisionChain(
current_uri="uko:v3",
predecessors=("uko:v1", "uko:v1"),
)
msg = "Expected ValueError for duplicate predecessors"
raise AssertionError(msg)
except ValidationError:
pass
# ---------------------------------------------------------------------------
# TierRetentionConfig steps
# ---------------------------------------------------------------------------
@given("a TierRetentionConfig with defaults")
def step_given_tier_retention_defaults(context: Any) -> None:
context.retention = TierRetentionConfig()
@then("the warm retention hours should be {expected:d}")
def step_then_retention_warm(context: Any, expected: int) -> None:
assert context.retention.warm_retention_hours == expected
@then("the cold retention days should be {expected:d}")
def step_then_retention_cold(context: Any, expected: int) -> None:
assert context.retention.cold_retention_days == expected
@then("modifying the retention warm hours should raise an error")
def step_then_retention_frozen(context: Any) -> None:
try:
context.retention.warm_retention_hours = 48 # type: ignore[misc]
msg = "Expected frozen model to reject mutation"
raise AssertionError(msg)
except ValidationError:
pass
@then(
"creating a TierRetentionConfig with warm_retention_hours 0 should raise ValueError"
)
def step_then_retention_zero_warm(context: Any) -> None:
try:
TierRetentionConfig(warm_retention_hours=0)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValidationError:
pass
@then(
"creating a TierRetentionConfig with cold_retention_days 0 should raise ValueError"
)
def step_then_retention_zero_cold(context: Any) -> None:
try:
TierRetentionConfig(cold_retention_days=0)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValidationError:
pass
@then('creating a revision from "{old}" to existing "{new}" should raise ValueError')
def step_then_create_revision_existing_uri_raises(
context: Any,
old: str,
new: str,
) -> None:
now = datetime.now(tz=UTC)
new_node = TemporalNode(
node_uri=new,
source_resource="RES001",
source_path="src/auth.py",
temporal=TemporalMetadata(
valid_from=now,
is_revision_of=old,
),
)
try:
context.temporal_backend.create_revision(old, new_node, now)
msg = "Expected ValueError for duplicate new_node URI"
raise AssertionError(msg)
except ValueError:
pass
# ---------------------------------------------------------------------------
# TierQueryResult steps
# ---------------------------------------------------------------------------
@given("a TierQueryResult for HOT tier with CURRENT scope")
def step_given_tier_query_result(context: Any) -> None:
context.tier_query_result = TierQueryResult(
tier=ContextTier.HOT,
temporal_scope=TemporalScope.CURRENT,
)
@then("the tier query result nodes should be empty")
def step_then_tier_query_nodes_empty(context: Any) -> None:
assert context.tier_query_result.nodes == ()
@then("the tier query result tier should be HOT")
def step_then_tier_query_tier_hot(context: Any) -> None:
assert context.tier_query_result.tier == ContextTier.HOT
@then("the tier query result temporal_scope should be CURRENT")
def step_then_tier_query_scope_current(context: Any) -> None:
assert context.tier_query_result.temporal_scope == TemporalScope.CURRENT
@then("modifying the tier query result tier should raise an error")
def step_then_tier_query_frozen(context: Any) -> None:
try:
context.tier_query_result.tier = ContextTier.WARM # type: ignore[misc]
msg = "Expected frozen model to reject mutation"
raise AssertionError(msg)
except ValidationError:
pass
# ---------------------------------------------------------------------------
# Structural assertion steps (Phase 3.1)
# ---------------------------------------------------------------------------
@then('TemporalMetadata should have field "{field}"')
def step_then_temporal_metadata_has_field(
context: Any,
field: str,
) -> None:
assert field in TemporalMetadata.model_fields
@then('TemporalNode should have field "{field}"')
def step_then_temporal_node_has_field(
context: Any,
field: str,
) -> None:
assert field in TemporalNode.model_fields
@then('RevisionChain should have field "{field}"')
def step_then_revision_chain_has_field(
context: Any,
field: str,
) -> None:
assert field in RevisionChain.model_fields
@then('TierRetentionConfig should have field "{field}"')
def step_then_retention_has_field(
context: Any,
field: str,
) -> None:
assert field in TierRetentionConfig.model_fields
@then('TierQueryResult should have field "{field}"')
def step_then_tier_query_has_field(
context: Any,
field: str,
) -> None:
assert field in TierQueryResult.model_fields
# ---------------------------------------------------------------------------
# StrategyConfig deep-freeze steps
# ---------------------------------------------------------------------------
@given("a StrategyConfig with nested extra dict")
def step_given_strategy_config_nested(context: Any) -> None:
context.strategy_config = StrategyConfig(
extra=MappingProxyType({"outer": {"inner": "value"}}),
)
@then("mutating the nested extra dict should raise TypeError")
def step_then_mutate_nested_extra_raises(context: Any) -> None:
try:
context.strategy_config.extra["outer"]["inner"] = "mutated"
msg = "Expected TypeError when mutating frozen nested dict"
raise AssertionError(msg)
except TypeError:
pass
# ---------------------------------------------------------------------------
# BackendSet steps (moved from backend_steps to stay under 500 lines)
# ---------------------------------------------------------------------------
@when("I create a BackendSet with the temporal backend")
def step_when_backendset_with_temporal(context: Any) -> None:
context.backend_set = BackendSet(
temporal=context.temporal_backend,
)
@then("the BackendSet temporal field should not be None")
def step_then_backendset_temporal_not_none(context: Any) -> None:
assert context.backend_set.temporal is not None
@then("it should satisfy the TemporalBackend protocol")
def step_then_satisfies_temporal_protocol(context: Any) -> None:
assert isinstance(context.temporal_backend, TemporalBackend)
# ---- Diff-coverage tier query steps (moved from backend_steps) ----
@when("I query the WARM tier")
def step_when_query_warm(context: Any) -> None:
retention = TierRetentionConfig()
context.tier_query = context.temporal_backend.query_by_tier(
ContextTier.WARM,
TemporalScope.ALL,
retention,
)
@when("I query the HOT tier with scope RECENT")
def step_when_query_hot_recent(context: Any) -> None:
retention = TierRetentionConfig()
context.tier_query = context.temporal_backend.query_by_tier(
ContextTier.HOT,
TemporalScope.RECENT,
retention,
)
# ---------------------------------------------------------------------------
# Round 4: fully-historical chain raises ValueError (NEW-2)
# ---------------------------------------------------------------------------
@given("a stored temporal node with all nodes historical")
def step_given_all_historical(context: Any) -> None:
now = datetime.now(tz=UTC)
v1_meta = TemporalMetadata(
valid_from=now - timedelta(days=2),
valid_until=now - timedelta(days=1),
is_current=False,
)
v2_meta = TemporalMetadata(
valid_from=now - timedelta(days=1),
valid_until=now,
is_current=False,
is_revision_of="uko-py:class/Dead_v1",
)
for uri, meta in [
("uko-py:class/Dead_v1", v1_meta),
("uko-py:class/Dead_v2", v2_meta),
]:
context.temporal_backend.store_node(
TemporalNode(
node_uri=uri,
source_resource="RES001",
source_path="src/dead.py",
temporal=meta,
)
)
@then("getting revision chain for fully-historical node should raise ValueError")
def step_then_historical_chain_raises(context: Any) -> None:
try:
context.temporal_backend.get_revision_chain("uko-py:class/Dead_v2")
msg = "Expected ValueError for fully-historical chain"
raise AssertionError(msg)
except ValueError:
pass
# ---------------------------------------------------------------------------
# Round 4: deep-freeze list → tuple (NEW-4)
# ---------------------------------------------------------------------------
@given("a StrategyConfig with nested list in extra")
def step_given_strategy_config_list(context: Any) -> None:
context.strategy_config = StrategyConfig(
extra=MappingProxyType({"items": [1, 2, 3]}),
)
@then("the nested list should be frozen as a tuple")
def step_then_list_frozen_as_tuple(context: Any) -> None:
val = context.strategy_config.extra["items"]
assert isinstance(val, tuple), f"Expected tuple, got {type(val).__name__}"
assert val == (1, 2, 3)
# ---------------------------------------------------------------------------
# Round 4: create_revision input validation (#14, #15)
# ---------------------------------------------------------------------------
@then(
'creating a revision from "{uri}" with wrong is_revision_of should raise ValueError'
)
def step_then_create_revision_wrong_revision_of(context: Any, uri: str) -> None:
now = datetime.now(tz=UTC)
new_node = TemporalNode(
node_uri="uko-py:class/Auth_v99",
source_resource="RES001",
source_path="src/auth.py",
temporal=TemporalMetadata(
valid_from=now,
is_revision_of="uko-py:class/WRONG",
),
)
try:
context.temporal_backend.create_revision(uri, new_node, now)
msg = "Expected ValueError for wrong is_revision_of"
raise AssertionError(msg)
except ValueError:
pass
@then('creating a revision from "{uri}" with is_current false should raise ValueError')
def step_then_create_revision_not_current(context: Any, uri: str) -> None:
now = datetime.now(tz=UTC)
new_node = TemporalNode(
node_uri="uko-py:class/Auth_v99",
source_resource="RES001",
source_path="src/auth.py",
temporal=TemporalMetadata(
valid_from=now,
valid_until=now + timedelta(days=1),
is_current=False,
is_revision_of=uri,
),
)
try:
context.temporal_backend.create_revision(uri, new_node, now)
msg = "Expected ValueError for is_current=False"
raise AssertionError(msg)
except ValueError:
pass
@@ -0,0 +1,431 @@
"""Step definitions for Temporal Data Model — metadata and node model steps."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta, timezone
from typing import Any
from behave import given, then
from pydantic import ValidationError
from cleveragents.domain.models.acms.temporal import (
TemporalMetadata,
TemporalNode,
)
__all__: list[str] = []
# ---------------------------------------------------------------------------
# TemporalMetadata steps
# ---------------------------------------------------------------------------
@given("a TemporalMetadata with valid_from now")
def step_given_temporal_metadata_defaults(context: Any) -> None:
context.temporal_meta = TemporalMetadata(
valid_from=datetime.now(tz=UTC),
)
@given(
"a TemporalMetadata with valid_from now and valid_until tomorrow"
' and is_current false and is_revision_of "{revision_of}"'
)
def step_given_temporal_metadata_all_fields(
context: Any,
revision_of: str,
) -> None:
now = datetime.now(tz=UTC)
context.temporal_meta = TemporalMetadata(
valid_from=now,
valid_until=now + timedelta(days=1),
is_current=False,
is_revision_of=revision_of,
)
@given("a TemporalMetadata with naive valid_from")
def step_given_temporal_metadata_naive_valid_from(context: Any) -> None:
context.temporal_meta = TemporalMetadata(
valid_from=datetime(2026, 1, 15, 10, 30, 0),
)
@given("a TemporalMetadata with naive valid_until")
def step_given_temporal_metadata_naive_valid_until(context: Any) -> None:
context.temporal_meta = TemporalMetadata(
valid_from=datetime.now(tz=UTC),
valid_until=datetime(2026, 12, 31, 23, 59, 59),
is_current=False,
)
@then("the temporal metadata is_current should be true")
def step_then_temporal_metadata_is_current_true(context: Any) -> None:
assert context.temporal_meta.is_current is True
@then("the temporal metadata is_current should be false")
def step_then_temporal_metadata_is_current_false(context: Any) -> None:
assert context.temporal_meta.is_current is False
@then("the temporal metadata valid_until should be None")
def step_then_temporal_metadata_valid_until_none(context: Any) -> None:
assert context.temporal_meta.valid_until is None
@then("the temporal metadata valid_until should not be None")
def step_then_temporal_metadata_valid_until_not_none(context: Any) -> None:
assert context.temporal_meta.valid_until is not None
@then("the temporal metadata is_revision_of should be None")
def step_then_temporal_metadata_revision_none(context: Any) -> None:
assert context.temporal_meta.is_revision_of is None
@then('the temporal metadata is_revision_of should be "{expected}"')
def step_then_temporal_metadata_revision_of(
context: Any,
expected: str,
) -> None:
assert context.temporal_meta.is_revision_of == expected
@then("the temporal metadata valid_from should have UTC timezone")
def step_then_temporal_metadata_valid_from_utc(context: Any) -> None:
assert context.temporal_meta.valid_from.tzinfo is not None
@then("the temporal metadata valid_until should have UTC timezone")
def step_then_temporal_metadata_valid_until_utc(context: Any) -> None:
assert context.temporal_meta.valid_until is not None
assert context.temporal_meta.valid_until.tzinfo is not None
@then("modifying the temporal metadata is_current should raise an error")
def step_then_temporal_metadata_frozen(context: Any) -> None:
try:
context.temporal_meta.is_current = False # type: ignore[misc]
msg = "Expected frozen model to reject mutation"
raise AssertionError(msg)
except ValidationError:
pass
@then("creating a TemporalMetadata with empty is_revision_of should raise ValueError")
def step_then_temporal_metadata_empty_revision(context: Any) -> None:
try:
TemporalMetadata(
valid_from=datetime.now(tz=UTC),
is_revision_of="",
)
msg = "Expected ValueError for empty is_revision_of"
raise AssertionError(msg)
except ValidationError:
pass
@then(
"creating a TemporalMetadata with whitespace is_revision_of should raise ValueError"
)
def step_then_temporal_metadata_whitespace_revision(context: Any) -> None:
try:
TemporalMetadata(
valid_from=datetime.now(tz=UTC),
is_revision_of=" ",
)
msg = "Expected ValueError for whitespace is_revision_of"
raise AssertionError(msg)
except ValidationError:
pass
@given("a TemporalMetadata with non-UTC timezone")
def step_given_temporal_metadata_non_utc(context: Any) -> None:
est = timezone(timedelta(hours=-5))
context.temporal_meta = TemporalMetadata(
valid_from=datetime(2026, 6, 15, 12, 0, tzinfo=est),
is_current=True,
)
@then(
"creating a TemporalMetadata with valid_until before valid_from"
" should raise ValueError",
)
def step_then_temporal_valid_until_before_valid_from(context: Any) -> None:
try:
now = datetime.now(tz=UTC)
TemporalMetadata(
valid_from=now + timedelta(hours=1),
valid_until=now,
is_current=False,
)
msg = "Expected ValueError for valid_until < valid_from"
raise AssertionError(msg)
except ValidationError:
pass
@then(
"creating a TemporalMetadata with is_current true and valid_until set"
" should raise ValueError",
)
def step_then_temporal_is_current_with_valid_until(context: Any) -> None:
try:
now = datetime.now(tz=UTC)
TemporalMetadata(
valid_from=now,
valid_until=now + timedelta(hours=1),
is_current=True,
)
msg = "Expected ValueError for is_current=True with valid_until"
raise AssertionError(msg)
except ValidationError:
pass
# ---------------------------------------------------------------------------
# TemporalNode steps
# ---------------------------------------------------------------------------
@given(
'a TemporalNode with uri "{uri}" and resource "{resource}"'
' and path "{path}" without range'
)
def step_given_temporal_node_no_range(
context: Any,
uri: str,
resource: str,
path: str,
) -> None:
context.temporal_node = TemporalNode(
node_uri=uri,
source_resource=resource,
source_path=path,
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
)
@given(
'a TemporalNode with uri "{uri}" and resource "{resource}"'
' and path "{path}" and range "{src_range}"'
)
def step_given_temporal_node_with_range(
context: Any,
uri: str,
resource: str,
path: str,
src_range: str,
) -> None:
context.temporal_node = TemporalNode(
node_uri=uri,
source_resource=resource,
source_path=path,
source_range=src_range,
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
)
@then('the temporal node uri should be "{expected}"')
def step_then_temporal_node_uri(context: Any, expected: str) -> None:
assert context.temporal_node.node_uri == expected
@then('the temporal node source_resource should be "{expected}"')
def step_then_temporal_node_resource(context: Any, expected: str) -> None:
assert context.temporal_node.source_resource == expected
@then('the temporal node source_path should be "{expected}"')
def step_then_temporal_node_path(context: Any, expected: str) -> None:
assert context.temporal_node.source_path == expected
@then("the temporal node source_range should be None")
def step_then_temporal_node_range_none(context: Any) -> None:
assert context.temporal_node.source_range is None
@then('the temporal node source_range should be "{expected}"')
def step_then_temporal_node_range(context: Any, expected: str) -> None:
assert context.temporal_node.source_range == expected
@then("the temporal node should have temporal metadata")
def step_then_temporal_node_has_metadata(context: Any) -> None:
assert isinstance(context.temporal_node.temporal, TemporalMetadata)
@then("modifying the temporal node uri should raise an error")
def step_then_temporal_node_frozen(context: Any) -> None:
try:
context.temporal_node.node_uri = "changed" # type: ignore[misc]
msg = "Expected frozen model to reject mutation"
raise AssertionError(msg)
except ValidationError:
pass
@then("creating a TemporalNode with empty node_uri should raise ValueError")
def step_then_temporal_node_empty_uri(context: Any) -> None:
try:
TemporalNode(
node_uri="",
source_resource="RES",
source_path="src/a.py",
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValidationError:
pass
@then("creating a TemporalNode with whitespace node_uri should raise ValueError")
def step_then_temporal_node_whitespace_uri(context: Any) -> None:
try:
TemporalNode(
node_uri=" ",
source_resource="RES",
source_path="src/a.py",
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValidationError:
pass
@then("creating a TemporalNode with empty source_resource should raise ValueError")
def step_then_temporal_node_empty_resource(context: Any) -> None:
try:
TemporalNode(
node_uri="uko:test",
source_resource="",
source_path="src/a.py",
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValidationError:
pass
@then("creating a TemporalNode with empty source_path should raise ValueError")
def step_then_temporal_node_empty_path(context: Any) -> None:
try:
TemporalNode(
node_uri="uko:test",
source_resource="RES",
source_path="",
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValidationError:
pass
@then("creating a TemporalNode with empty source_range should raise ValueError")
def step_then_temporal_node_empty_range(context: Any) -> None:
try:
TemporalNode(
node_uri="uko:test",
source_resource="RES",
source_path="src/a.py",
source_range="",
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValidationError:
pass
@then("creating a TemporalNode with whitespace source_range should raise ValueError")
def step_then_temporal_node_whitespace_range(context: Any) -> None:
try:
TemporalNode(
node_uri="uko:test",
source_resource="RES",
source_path="src/a.py",
source_range=" ",
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValidationError:
pass
# ---------------------------------------------------------------------------
# Round 4: string datetime coercion (NEW-3)
# ---------------------------------------------------------------------------
@given("a TemporalMetadata with valid_from as ISO-8601 string")
def step_given_metadata_string_datetime(context: Any) -> None:
context.temporal_meta = TemporalMetadata.model_validate(
{
"valid_from": "2025-06-15T12:00:00+03:00",
"is_current": True,
}
)
# ---------------------------------------------------------------------------
# Round 5: zero-width Unicode (NEW-14) + symmetric invariant (NEW-15)
# ---------------------------------------------------------------------------
@then(
"creating a TemporalNode with zero-width Unicode node_uri should raise ValueError"
)
def step_then_zero_width_uri_raises(context: Any) -> None:
try:
TemporalNode(
node_uri="\u200b\u200c\u200d",
source_resource="RES001",
source_path="src/a.py",
temporal=TemporalMetadata(valid_from=datetime.now(tz=UTC)),
)
msg = "Expected ValueError for zero-width Unicode URI"
raise AssertionError(msg)
except ValidationError:
pass
@then(
"creating a TemporalMetadata with is_current false"
" and no valid_until should raise ValueError"
)
def step_then_historical_without_valid_until_raises(context: Any) -> None:
try:
TemporalMetadata(
valid_from=datetime.now(tz=UTC),
is_current=False,
)
msg = "Expected ValueError for is_current=False without valid_until"
raise AssertionError(msg)
except ValidationError:
pass
@then(
"creating a TemporalMetadata with zero-width Unicode is_revision_of"
" should raise ValueError"
)
def step_then_zero_width_revision_of_raises(context: Any) -> None:
try:
TemporalMetadata(
valid_from=datetime.now(tz=UTC),
is_revision_of="\u200b\u200c",
)
msg = "Expected ValueError for zero-width Unicode is_revision_of"
raise AssertionError(msg)
except ValidationError:
pass
@@ -0,0 +1,321 @@
"""Step definitions for Temporal Data Model — service steps."""
from __future__ import annotations
from typing import Any
from behave import given, then, when
from cleveragents.application.services.temporal_service import TemporalService
from cleveragents.domain.models.acms.temporal_stubs import InMemoryTemporalBackend
from cleveragents.domain.models.acms.tiers import ContextTier
from cleveragents.domain.models.core.project import TemporalScope
__all__: list[str] = []
# ---------------------------------------------------------------------------
# TemporalService steps
# ---------------------------------------------------------------------------
@given("a TemporalService with InMemoryTemporalBackend")
def step_given_temporal_service(context: Any) -> None:
backend = InMemoryTemporalBackend()
context.temporal_service = TemporalService(backend=backend)
context.temporal_backend = backend
@given('I store an initial node "{uri}" for resource "{resource}" at "{path}"')
@when('I store an initial node "{uri}" for resource "{resource}" at "{path}"')
def step_when_store_initial(
context: Any,
uri: str,
resource: str,
path: str,
) -> None:
context.service_node = context.temporal_service.store_initial_node(
node_uri=uri,
source_resource=resource,
source_path=path,
)
@then("the stored node should be current")
def step_then_stored_current(context: Any) -> None:
assert context.service_node.temporal.is_current is True
@then("the stored node should have no predecessor")
def step_then_stored_no_predecessor(context: Any) -> None:
assert context.service_node.temporal.is_revision_of is None
@when(
'I create a service revision from "{old}" to "{new}"'
' for resource "{resource}" at "{path}"'
)
def step_when_service_create_revision(
context: Any,
old: str,
new: str,
resource: str,
path: str,
) -> None:
context.service_revision = context.temporal_service.create_revision(
current_uri=old,
new_node_uri=new,
source_resource=resource,
source_path=path,
)
@then("the service revision node should be current")
def step_then_service_revision_current(context: Any) -> None:
assert context.service_revision.temporal.is_current is True
@then('the service revision node should have is_revision_of "{expected}"')
def step_then_service_revision_of(context: Any, expected: str) -> None:
assert context.service_revision.temporal.is_revision_of == expected
@when('I get the service current for "{base}"')
def step_when_service_get_current(context: Any, base: str) -> None:
context.service_current = context.temporal_service.get_current(base)
@then('the service current node uri should be "{expected}"')
def step_then_service_current_uri(context: Any, expected: str) -> None:
assert context.service_current is not None
assert context.service_current.node_uri == expected
@when("I query HOT tier via the service")
def step_when_service_query_hot(context: Any) -> None:
context.service_tier_query = context.temporal_service.query_by_tier(
ContextTier.HOT,
)
@when("I query HOT tier via the service with scope ALL")
def step_when_service_query_hot_all(context: Any) -> None:
context.service_tier_query = context.temporal_service.query_by_tier(
ContextTier.HOT,
temporal_scope=TemporalScope.ALL,
)
@then("the service tier query scope should be CURRENT")
def step_then_service_tier_scope_current(context: Any) -> None:
assert context.service_tier_query.temporal_scope == TemporalScope.CURRENT
@then("the service tier query scope should be ALL")
def step_then_service_tier_scope_all(context: Any) -> None:
assert context.service_tier_query.temporal_scope == TemporalScope.ALL
@when('I mark "{uri}" as historical via the service')
def step_when_service_mark_historical(context: Any, uri: str) -> None:
context.service_marked = context.temporal_service.mark_historical(uri)
@then("the service marked node should not be current")
def step_then_service_marked_not_current(context: Any) -> None:
assert context.service_marked.temporal.is_current is False
@then("storing an initial node with blank URI via the service should raise ValueError")
def step_then_service_blank_uri(context: Any) -> None:
try:
context.temporal_service.store_initial_node(
node_uri="",
source_resource="RES",
source_path="src/a.py",
)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValueError:
pass
@then(
"storing an initial node with blank source_resource via the service"
" should raise ValueError"
)
def step_then_service_blank_resource(context: Any) -> None:
try:
context.temporal_service.store_initial_node(
node_uri="uko:test",
source_resource="",
source_path="src/a.py",
)
msg = "Expected ValueError"
raise AssertionError(msg)
except ValueError:
pass
@when('I get the service history for "{base}" with scope ALL')
def step_when_service_get_history_all(context: Any, base: str) -> None:
context.service_history = context.temporal_service.get_history(
base,
temporal_scope=TemporalScope.ALL,
)
@when('I get the service history for "{base}" with scope CURRENT')
def step_when_service_get_history_current(context: Any, base: str) -> None:
context.service_history = context.temporal_service.get_history(
base,
temporal_scope=TemporalScope.CURRENT,
)
@then("the service history should have {count:d} nodes")
def step_then_service_history_count(context: Any, count: int) -> None:
assert len(context.service_history) == count, (
f"Expected {count} nodes, got {len(context.service_history)}"
)
@then("getting service history for blank base URI should raise ValueError")
def step_then_service_history_blank_raises(context: Any) -> None:
try:
context.temporal_service.get_history("")
msg = "Expected ValueError"
raise AssertionError(msg)
except ValueError:
pass
@when('I get the service revision chain for "{uri}"')
def step_when_service_get_chain(context: Any, uri: str) -> None:
context.service_chain = context.temporal_service.get_revision_chain(uri)
@then("the service revision chain depth should be {expected:d}")
def step_then_service_chain_depth(context: Any, expected: int) -> None:
assert context.service_chain.depth == expected, (
f"Expected depth {expected}, got {context.service_chain.depth}"
)
@then('the service revision chain current should be "{expected}"')
def step_then_service_chain_current(context: Any, expected: str) -> None:
assert context.service_chain.current_uri == expected, (
f"Expected {expected}, got {context.service_chain.current_uri}"
)
@then("getting service revision chain for blank URI should raise ValueError")
def step_then_service_chain_blank_raises(context: Any) -> None:
try:
context.temporal_service.get_revision_chain("")
msg = "Expected ValueError"
raise AssertionError(msg)
except ValueError:
pass
@when("I query COLD tier via the service with scope CURRENT")
def step_when_service_query_cold_current(context: Any) -> None:
context.service_tier_query = context.temporal_service.query_by_tier(
ContextTier.COLD,
temporal_scope=TemporalScope.CURRENT,
)
@then("the service tier query should return {count:d} node")
def step_then_service_tier_query_count(context: Any, count: int) -> None:
assert len(context.service_tier_query.nodes) == count, (
f"Expected {count} nodes, got {len(context.service_tier_query.nodes)}"
)
@then("the service tier query node should be current")
def step_then_service_tier_query_node_current(context: Any) -> None:
for node in context.service_tier_query.nodes:
assert node.temporal.is_current is True, f"Node {node.node_uri} is not current"
@then("the service retention warm hours should be {expected:d}")
def step_then_service_retention_warm(context: Any, expected: int) -> None:
assert context.temporal_service.retention.warm_retention_hours == expected
@then("the service retention cold days should be {expected:d}")
def step_then_service_retention_cold(context: Any, expected: int) -> None:
assert context.temporal_service.retention.cold_retention_days == expected
# ---------------------------------------------------------------------------
# Diff-coverage steps (WARM/COLD default scopes)
# ---------------------------------------------------------------------------
@when("I query WARM tier via the service")
def step_when_service_query_warm(context: Any) -> None:
context.service_tier_query = context.temporal_service.query_by_tier(
ContextTier.WARM,
)
@when("I query COLD tier via the service")
def step_when_service_query_cold(context: Any) -> None:
context.service_tier_query = context.temporal_service.query_by_tier(
ContextTier.COLD,
)
@then("the service tier query scope should be RECENT")
def step_then_service_tier_scope_recent(context: Any) -> None:
assert context.service_tier_query.temporal_scope == TemporalScope.RECENT
# ---------------------------------------------------------------------------
# Edge-case guards (Round 3 bug-hunt fixes)
# ---------------------------------------------------------------------------
@then(
'creating a service revision from "{uri}" to itself should raise ValueError',
)
def step_then_service_same_uri_raises(context: Any, uri: str) -> None:
try:
context.temporal_service.create_revision(
current_uri=uri,
new_node_uri=uri,
source_resource="RES001",
source_path="src/foo.py",
)
msg = "Expected ValueError for same-URI service revision"
raise AssertionError(msg)
except ValueError:
pass
@then(
'marking "{uri}" as historical again via the service should raise ValueError',
)
def step_then_service_mark_historical_again_raises(
context: Any,
uri: str,
) -> None:
try:
context.temporal_service.mark_historical(uri)
msg = "Expected ValueError for already-historical node"
raise AssertionError(msg)
except ValueError:
pass
@then("creating a TemporalService with None backend should raise TypeError")
def step_then_service_none_backend(context: Any) -> None:
try:
TemporalService(backend=None) # type: ignore[arg-type]
msg = "Expected TypeError for None backend"
raise AssertionError(msg)
except TypeError:
pass
+496
View File
@@ -0,0 +1,496 @@
Feature: Temporal Data Model (Revision-Aware RDF) with 3 Storage Tiers
As a developer
I want temporal metadata on UKO nodes with revision chains and tier-aware queries
So that the ACMS can track knowledge graph evolution and enable temporal-archaeology
# ---- TemporalMetadata model ----
Scenario: Create a valid TemporalMetadata with defaults
Given a TemporalMetadata with valid_from now
Then the temporal metadata is_current should be true
And the temporal metadata valid_until should be None
And the temporal metadata is_revision_of should be None
And the temporal metadata valid_from should have UTC timezone
Scenario: Create a TemporalMetadata with all fields
Given a TemporalMetadata with valid_from now and valid_until tomorrow and is_current false and is_revision_of "uko-py:class/Foo_v1"
Then the temporal metadata is_current should be false
And the temporal metadata valid_until should not be None
And the temporal metadata is_revision_of should be "uko-py:class/Foo_v1"
Scenario: TemporalMetadata is frozen
Given a TemporalMetadata with valid_from now
Then modifying the temporal metadata is_current should raise an error
Scenario: TemporalMetadata rejects empty is_revision_of
Then creating a TemporalMetadata with empty is_revision_of should raise ValueError
Scenario: TemporalMetadata rejects whitespace-only is_revision_of
Then creating a TemporalMetadata with whitespace is_revision_of should raise ValueError
Scenario: TemporalNode rejects zero-width Unicode node_uri
Then creating a TemporalNode with zero-width Unicode node_uri should raise ValueError
Scenario: TemporalMetadata rejects zero-width Unicode is_revision_of
Then creating a TemporalMetadata with zero-width Unicode is_revision_of should raise ValueError
Scenario: TemporalMetadata adds UTC to naive valid_from
Given a TemporalMetadata with naive valid_from
Then the temporal metadata valid_from should have UTC timezone
Scenario: TemporalMetadata adds UTC to naive valid_until
Given a TemporalMetadata with naive valid_until
Then the temporal metadata valid_until should have UTC timezone
Scenario: TemporalMetadata converts non-UTC aware datetime to UTC
Given a TemporalMetadata with non-UTC timezone
Then the temporal metadata valid_from should have UTC timezone
Scenario: TemporalMetadata coerces ISO-8601 string datetime to UTC
Given a TemporalMetadata with valid_from as ISO-8601 string
Then the temporal metadata valid_from should have UTC timezone
Scenario: TemporalMetadata rejects valid_until before valid_from
Then creating a TemporalMetadata with valid_until before valid_from should raise ValueError
Scenario: TemporalMetadata rejects is_current with valid_until set
Then creating a TemporalMetadata with is_current true and valid_until set should raise ValueError
Scenario: TemporalMetadata rejects is_current false without valid_until
Then creating a TemporalMetadata with is_current false and no valid_until should raise ValueError
# ---- TemporalNode model ----
Scenario: Create a valid TemporalNode
Given a TemporalNode with uri "uko-py:class/Auth_v1" and resource "RES001" and path "src/auth.py" without range
Then the temporal node uri should be "uko-py:class/Auth_v1"
And the temporal node source_resource should be "RES001"
And the temporal node source_path should be "src/auth.py"
And the temporal node source_range should be None
And the temporal node should have temporal metadata
Scenario: TemporalNode with source_range
Given a TemporalNode with uri "uko-py:class/Auth_v1" and resource "RES001" and path "src/auth.py" and range "15:1-87:0"
Then the temporal node source_range should be "15:1-87:0"
Scenario: TemporalNode is frozen
Given a TemporalNode with uri "uko-py:class/Auth_v1" and resource "RES001" and path "src/auth.py" without range
Then modifying the temporal node uri should raise an error
Scenario: TemporalNode rejects empty node_uri
Then creating a TemporalNode with empty node_uri should raise ValueError
Scenario: TemporalNode rejects whitespace-only node_uri
Then creating a TemporalNode with whitespace node_uri should raise ValueError
Scenario: TemporalNode rejects empty source_resource
Then creating a TemporalNode with empty source_resource should raise ValueError
Scenario: TemporalNode rejects empty source_path
Then creating a TemporalNode with empty source_path should raise ValueError
Scenario: TemporalNode rejects empty source_range
Then creating a TemporalNode with empty source_range should raise ValueError
Scenario: TemporalNode rejects whitespace-only source_range
Then creating a TemporalNode with whitespace source_range should raise ValueError
# ---- RevisionChain model ----
Scenario: Create a RevisionChain with no predecessors
Given a RevisionChain with only current_uri "uko-py:class/Auth_v1"
Then the revision chain depth should be 1
And the revision chain predecessors should be empty
And the revision chain all_uris should contain "uko-py:class/Auth_v1"
Scenario: Create a RevisionChain with predecessors
Given a RevisionChain with current_uri "uko-py:class/Auth_v3" and predecessors "uko-py:class/Auth_v1,uko-py:class/Auth_v2"
Then the revision chain depth should be 3
And the revision chain predecessors should have 2 entries
And the revision chain all_uris should have 3 entries
Scenario: RevisionChain is frozen
Given a RevisionChain with only current_uri "uko-py:class/Auth_v1"
Then modifying the revision chain current_uri should raise an error
Scenario: RevisionChain rejects empty current_uri
Then creating a RevisionChain with empty current_uri should raise ValueError
Scenario: RevisionChain rejects whitespace-only current_uri
Then creating a RevisionChain with whitespace current_uri should raise ValueError
Scenario: RevisionChain rejects empty predecessor URI
Then creating a RevisionChain with empty predecessor should raise ValueError
Scenario: RevisionChain rejects current_uri in predecessors
Then creating a RevisionChain with current_uri in predecessors should raise ValueError
Scenario: RevisionChain rejects duplicate predecessors
Then creating a RevisionChain with duplicate predecessors should raise ValueError
# ---- TierRetentionConfig model ----
Scenario: TierRetentionConfig defaults match spec
Given a TierRetentionConfig with defaults
Then the warm retention hours should be 24
And the cold retention days should be 90
Scenario: TierRetentionConfig is frozen
Given a TierRetentionConfig with defaults
Then modifying the retention warm hours should raise an error
Scenario: TierRetentionConfig rejects zero warm hours
Then creating a TierRetentionConfig with warm_retention_hours 0 should raise ValueError
Scenario: TierRetentionConfig rejects zero cold days
Then creating a TierRetentionConfig with cold_retention_days 0 should raise ValueError
# ---- TierQueryResult model ----
Scenario: Create a TierQueryResult with no nodes
Given a TierQueryResult for HOT tier with CURRENT scope
Then the tier query result nodes should be empty
And the tier query result tier should be HOT
And the tier query result temporal_scope should be CURRENT
Scenario: TierQueryResult is frozen
Given a TierQueryResult for HOT tier with CURRENT scope
Then modifying the tier query result tier should raise an error
# ---- TemporalBackend protocol ----
Scenario: InMemoryTemporalBackend satisfies TemporalBackend protocol
Given an InMemoryTemporalBackend instance
Then it should satisfy the TemporalBackend protocol
# ---- InMemoryTemporalBackend: store and get ----
Scenario: Store a node and get current
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
When I get the current node for base "uko-py:class/Auth"
Then the current node uri should be "uko-py:class/Auth_v1"
Scenario: Get current returns None when no match
Given an InMemoryTemporalBackend instance
When I get the current node for base "uko-py:class/NonExistent"
Then the current node should be None
Scenario: Get current does not match unrelated URI prefix
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/AuthManager_v1" for resource "RES001" at "src/auth.py"
When I get the current node for base "uko-py:class/Auth"
Then the current node should be None
# ---- InMemoryTemporalBackend: create revision ----
Scenario: Create a revision marks old node historical
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
When I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
Then the old node "uko-py:class/Auth_v1" should not be current
And the old node "uko-py:class/Auth_v1" should have valid_until set
And the new node "uko-py:class/Auth_v2" should be current
And the new node "uko-py:class/Auth_v2" should have is_revision_of "uko-py:class/Auth_v1"
Scenario: Create revision from nonexistent URI raises KeyError
Given an InMemoryTemporalBackend instance
Then creating a revision from nonexistent "uko-py:class/Missing" should raise KeyError
Scenario: Create revision with blank URI raises ValueError
Given an InMemoryTemporalBackend instance
Then creating a revision from blank URI should raise ValueError
Scenario: Create revision with same URI as current raises ValueError
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
Then creating a revision from "uko-py:class/Auth_v1" to itself should raise ValueError
Scenario: Store node rejects duplicate URI
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
Then storing a duplicate node "uko-py:class/Auth_v1" should raise ValueError
Scenario: Create revision rejects URI that already exists
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And a stored temporal node "uko-py:class/Auth_v2" for resource "RES001" at "src/auth.py"
Then creating a revision from "uko-py:class/Auth_v1" to existing "uko-py:class/Auth_v2" should raise ValueError
Scenario: Create revision rejects wrong is_revision_of on new node
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
Then creating a revision from "uko-py:class/Auth_v1" with wrong is_revision_of should raise ValueError
Scenario: Create revision rejects is_current false on new node
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
Then creating a revision from "uko-py:class/Auth_v1" with is_current false should raise ValueError
# ---- InMemoryTemporalBackend: get history ----
Scenario: Get history with ALL scope returns all versions
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
When I get the history for base "uko-py:class/Auth" with scope ALL
Then the history should have 2 nodes
And the history should be ordered newest first
Scenario: Get history with CURRENT scope returns only current
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
When I get the history for base "uko-py:class/Auth" with scope CURRENT
Then the history should have 1 nodes
Scenario: Get history with blank base URI raises ValueError
Given an InMemoryTemporalBackend instance
Then getting history for blank base URI should raise ValueError
# ---- InMemoryTemporalBackend: revision chain ----
Scenario: Get revision chain for single node
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
When I get the revision chain for "uko-py:class/Auth_v1"
Then the revision chain should have depth 1
And the revision chain current should be "uko-py:class/Auth_v1"
Scenario: Get revision chain after multiple revisions
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
And I create a revision from "uko-py:class/Auth_v2" to "uko-py:class/Auth_v3"
When I get the revision chain for "uko-py:class/Auth_v3"
Then the revision chain should have depth 3
And the revision chain current should be "uko-py:class/Auth_v3"
And the revision chain predecessors should include "uko-py:class/Auth_v1"
And the revision chain predecessors should include "uko-py:class/Auth_v2"
Scenario: Get revision chain from middle node includes full chain
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
And I create a revision from "uko-py:class/Auth_v2" to "uko-py:class/Auth_v3"
When I get the revision chain for "uko-py:class/Auth_v1"
Then the revision chain should have depth 3
Scenario: Get revision chain for nonexistent URI raises KeyError
Given an InMemoryTemporalBackend instance
Then getting revision chain for nonexistent URI should raise KeyError
Scenario: Fully-historical revision chain raises ValueError
Given an InMemoryTemporalBackend instance
And a stored temporal node with all nodes historical
Then getting revision chain for fully-historical node should raise ValueError
# ---- InMemoryTemporalBackend: query by tier ----
Scenario: Hot tier query returns only current nodes
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
When I query the HOT tier
Then the tier query should return 1 node
And the tier query node should be current
Scenario: Cold tier query returns all temporal versions
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
When I query the COLD tier
Then the tier query should return 2 nodes
# ---- InMemoryTemporalBackend: mark historical ----
Scenario: Mark a node as historical
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
When I mark "uko-py:class/Auth_v1" as historical
Then the marked node should not be current
And the marked node should have valid_until set
Scenario: Mark historical with blank URI raises ValueError
Given an InMemoryTemporalBackend instance
Then marking blank URI as historical should raise ValueError
Scenario: Mark historical with nonexistent URI raises KeyError
Given an InMemoryTemporalBackend instance
Then marking nonexistent URI as historical should raise KeyError
Scenario: Mark historical on already-historical node raises ValueError
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I mark "uko-py:class/Auth_v1" as historical
Then marking already-historical "uko-py:class/Auth_v1" should raise ValueError
Scenario: Create revision on already-historical node raises ValueError
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
Then creating a revision from historical "uko-py:class/Auth_v1" to "uko-py:class/Auth_v3" should raise ValueError
Scenario: Get revision chain identifies current head explicitly
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
When I get the revision chain for "uko-py:class/Auth_v1"
Then the revision chain current_uri should be "uko-py:class/Auth_v2"
And the revision chain predecessors should contain "uko-py:class/Auth_v1"
Scenario: Get revision chain excludes sibling branch nodes
Given an InMemoryTemporalBackend instance
And a stored temporal node with branching revision chain
When I get the revision chain for "uko-py:class/Branch_B"
Then the revision chain should not contain "uko-py:class/Branch_C"
And the revision chain current_uri should be "uko-py:class/Branch_B"
# ---- InMemoryTemporalBackend: diff-coverage scenarios ----
Scenario: Get history with RECENT scope returns current and recently expired nodes
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
When I get the history for base "uko-py:class/Auth" with scope RECENT
Then the history should have 2 nodes
Scenario: Revision chain with cycle guard stops traversal
Given an InMemoryTemporalBackend instance
And a stored temporal node with cyclic revision chain
When I get the revision chain for "uko-py:class/Cycle_B"
Then the revision chain should have depth 2
Scenario: Revision chain with missing predecessor stops traversal
Given an InMemoryTemporalBackend instance
And a stored temporal node with dangling predecessor
When I get the revision chain for "uko-py:class/Dangling_B"
Then the revision chain should have depth 1
Scenario: Warm tier query returns current and recently expired nodes
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
When I query the WARM tier
Then the tier query should return 2 nodes
Scenario: Query by tier with RECENT scope filters within tier pool
Given an InMemoryTemporalBackend instance
And a stored temporal node "uko-py:class/Auth_v1" for resource "RES001" at "src/auth.py"
And I create a revision from "uko-py:class/Auth_v1" to "uko-py:class/Auth_v2"
When I query the HOT tier with scope RECENT
Then the tier query should return 1 node
# ---- TemporalService ----
Scenario: TemporalService store initial node
Given a TemporalService with InMemoryTemporalBackend
When I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
Then the stored node should be current
And the stored node should have no predecessor
Scenario: TemporalService create revision
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I create a service revision from "uko-py:class/Foo_v1" to "uko-py:class/Foo_v2" for resource "RES001" at "src/foo.py"
Then the service revision node should be current
And the service revision node should have is_revision_of "uko-py:class/Foo_v1"
Scenario: TemporalService get current
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I get the service current for "uko-py:class/Foo"
Then the service current node uri should be "uko-py:class/Foo_v1"
Scenario: TemporalService query by tier uses default scope
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I query HOT tier via the service
Then the service tier query scope should be CURRENT
Scenario: TemporalService query by tier allows scope override
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I query HOT tier via the service with scope ALL
Then the service tier query scope should be ALL
Scenario: TemporalService mark historical
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I mark "uko-py:class/Foo_v1" as historical via the service
Then the service marked node should not be current
Scenario: TemporalService rejects None backend
Then creating a TemporalService with None backend should raise TypeError
Scenario: TemporalService rejects blank node_uri
Given a TemporalService with InMemoryTemporalBackend
Then storing an initial node with blank URI via the service should raise ValueError
Scenario: TemporalService rejects blank source_resource
Given a TemporalService with InMemoryTemporalBackend
Then storing an initial node with blank source_resource via the service should raise ValueError
Scenario: TemporalService get history returns matching versions
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I create a service revision from "uko-py:class/Foo_v1" to "uko-py:class/Foo_v2" for resource "RES001" at "src/foo.py"
And I get the service history for "uko-py:class/Foo" with scope ALL
Then the service history should have 2 nodes
Scenario: TemporalService get history with CURRENT scope returns only current
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I create a service revision from "uko-py:class/Foo_v1" to "uko-py:class/Foo_v2" for resource "RES001" at "src/foo.py"
And I get the service history for "uko-py:class/Foo" with scope CURRENT
Then the service history should have 1 nodes
Scenario: TemporalService get history rejects blank base URI
Given a TemporalService with InMemoryTemporalBackend
Then getting service history for blank base URI should raise ValueError
Scenario: TemporalService get revision chain returns full chain
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I create a service revision from "uko-py:class/Foo_v1" to "uko-py:class/Foo_v2" for resource "RES001" at "src/foo.py"
And I get the service revision chain for "uko-py:class/Foo_v2"
Then the service revision chain depth should be 2
And the service revision chain current should be "uko-py:class/Foo_v2"
Scenario: TemporalService get revision chain rejects blank URI
Given a TemporalService with InMemoryTemporalBackend
Then getting service revision chain for blank URI should raise ValueError
Scenario: TemporalService query COLD with CURRENT scope returns only current
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I create a service revision from "uko-py:class/Foo_v1" to "uko-py:class/Foo_v2" for resource "RES001" at "src/foo.py"
And I query COLD tier via the service with scope CURRENT
Then the service tier query should return 1 node
And the service tier query node should be current
Scenario: TemporalService default retention matches spec
Given a TemporalService with InMemoryTemporalBackend
Then the service retention warm hours should be 24
And the service retention cold days should be 90
Scenario: TemporalService WARM tier uses RECENT default scope
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I query WARM tier via the service
Then the service tier query scope should be RECENT
Scenario: TemporalService COLD tier uses ALL default scope
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I query COLD tier via the service
Then the service tier query scope should be ALL
Scenario: TemporalService create revision with same URI raises ValueError
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
Then creating a service revision from "uko-py:class/Foo_v1" to itself should raise ValueError
Scenario: TemporalService mark historical on already-historical raises ValueError
Given a TemporalService with InMemoryTemporalBackend
And I store an initial node "uko-py:class/Foo_v1" for resource "RES001" at "src/foo.py"
When I mark "uko-py:class/Foo_v1" as historical via the service
Then marking "uko-py:class/Foo_v1" as historical again via the service should raise ValueError
# ---- BackendSet temporal typing ----
Scenario: BackendSet accepts TemporalBackend
Given an InMemoryTemporalBackend instance
When I create a BackendSet with the temporal backend
Then the BackendSet temporal field should not be None
Scenario: StrategyConfig extra deep-freezes nested dicts
Given a StrategyConfig with nested extra dict
Then mutating the nested extra dict should raise TypeError
Scenario: StrategyConfig extra deep-freezes nested lists to tuples
Given a StrategyConfig with nested list in extra
Then the nested list should be frozen as a tuple
# ---- Structural assertions (Phase 3.1) ----
Scenario: TemporalMetadata has all spec-required fields
Then TemporalMetadata should have field "valid_from"
And TemporalMetadata should have field "valid_until"
And TemporalMetadata should have field "is_current"
And TemporalMetadata should have field "is_revision_of"
Scenario: TemporalNode has all spec-required fields
Then TemporalNode should have field "node_uri"
And TemporalNode should have field "source_resource"
And TemporalNode should have field "source_path"
And TemporalNode should have field "source_range"
And TemporalNode should have field "temporal"
Scenario: RevisionChain has all spec-required fields
Then RevisionChain should have field "current_uri"
And RevisionChain should have field "predecessors"
Scenario: TierRetentionConfig has all spec-required fields
Then TierRetentionConfig should have field "warm_retention_hours"
And TierRetentionConfig should have field "cold_retention_days"
Scenario: TierQueryResult has all spec-required fields
Then TierQueryResult should have field "nodes"
And TierQueryResult should have field "tier"
And TierQueryResult should have field "temporal_scope"
+363
View File
@@ -0,0 +1,363 @@
"""Robot Framework helper for Temporal Data Model smoke tests.
Provides a CLI-style interface for Robot to invoke temporal model
creation, revision chain operations, tier queries, and service-level
operations. Exit code 0 = success, 1 = failure.
Usage:
python robot/helper_temporal_data_model.py <command>
"""
from __future__ import annotations
import sys
from datetime import UTC, datetime, timedelta
from pathlib import Path
# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.application.services.temporal_service import ( # noqa: E402
TemporalService,
)
from cleveragents.domain.models.acms.temporal import ( # noqa: E402
RevisionChain,
TemporalBackend,
TemporalMetadata,
TemporalNode,
TierQueryResult,
TierRetentionConfig,
)
from cleveragents.domain.models.acms.temporal_stubs import ( # noqa: E402
InMemoryTemporalBackend,
)
from cleveragents.domain.models.acms.tiers import ContextTier # noqa: E402
from cleveragents.domain.models.core.project import ( # noqa: E402
TemporalScope,
)
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print("Usage: helper_temporal_data_model.py <command>")
return 1
command: str = sys.argv[1]
if command == "protocol-compliance":
return _test_protocol_compliance()
if command == "revision-chain":
return _test_revision_chain()
if command == "tier-queries":
return _test_tier_queries()
if command == "service-lifecycle":
return _test_service_lifecycle()
if command == "model-creation":
return _test_model_creation()
if command == "validation":
return _test_validation()
if command == "retention-defaults":
return _test_retention_defaults()
if command == "integration-modify-resource":
return _test_integration_modify_resource()
print(f"Unknown command: {command}")
return 1
def _test_protocol_compliance() -> int:
"""Verify InMemoryTemporalBackend satisfies TemporalBackend."""
try:
backend = InMemoryTemporalBackend()
assert isinstance(backend, TemporalBackend)
print("temporal-protocol-ok")
return 0
except Exception as exc:
print(f"FAIL: {exc}")
return 1
def _test_revision_chain() -> int:
"""Test revision chain creation and traversal."""
try:
backend = InMemoryTemporalBackend()
now = datetime.now(tz=UTC)
# Store v1
v1 = TemporalNode(
node_uri="uko-py:class/Auth_v1",
source_resource="RES001",
source_path="src/auth.py",
temporal=TemporalMetadata(
valid_from=now - timedelta(days=5),
is_current=True,
),
)
backend.store_node(v1)
# Create v2
v2 = TemporalNode(
node_uri="uko-py:class/Auth_v2",
source_resource="RES001",
source_path="src/auth.py",
temporal=TemporalMetadata(
valid_from=now - timedelta(days=2),
is_current=True,
is_revision_of="uko-py:class/Auth_v1",
),
)
backend.create_revision("uko-py:class/Auth_v1", v2, now - timedelta(days=2))
# Create v3
v3 = TemporalNode(
node_uri="uko-py:class/Auth_v3",
source_resource="RES001",
source_path="src/auth.py",
temporal=TemporalMetadata(
valid_from=now,
is_current=True,
is_revision_of="uko-py:class/Auth_v2",
),
)
backend.create_revision("uko-py:class/Auth_v2", v3, now)
# Verify chain
chain = backend.get_revision_chain("uko-py:class/Auth_v3")
assert chain.depth == 3
assert chain.current_uri == "uko-py:class/Auth_v3"
assert "uko-py:class/Auth_v1" in chain.predecessors
assert "uko-py:class/Auth_v2" in chain.predecessors
print("temporal-revision-chain-ok")
return 0
except Exception as exc:
print(f"FAIL: {exc}")
return 1
def _test_tier_queries() -> int:
"""Test tier-aware temporal queries."""
try:
backend = InMemoryTemporalBackend()
now = datetime.now(tz=UTC)
retention = TierRetentionConfig()
# Store a current node and a historical node
v1 = TemporalNode(
node_uri="uko-py:class/Foo_v1",
source_resource="RES001",
source_path="src/foo.py",
temporal=TemporalMetadata(
valid_from=now - timedelta(days=5),
is_current=True,
),
)
backend.store_node(v1)
v2 = TemporalNode(
node_uri="uko-py:class/Foo_v2",
source_resource="RES001",
source_path="src/foo.py",
temporal=TemporalMetadata(
valid_from=now,
is_current=True,
is_revision_of="uko-py:class/Foo_v1",
),
)
backend.create_revision("uko-py:class/Foo_v1", v2, now)
# Hot tier: only current
hot = backend.query_by_tier(ContextTier.HOT, TemporalScope.CURRENT, retention)
assert len(hot.nodes) == 1
assert hot.nodes[0].temporal.is_current is True
# Cold tier: all
cold = backend.query_by_tier(ContextTier.COLD, TemporalScope.ALL, retention)
assert len(cold.nodes) == 2
print("temporal-tier-queries-ok")
return 0
except Exception as exc:
print(f"FAIL: {exc}")
return 1
def _test_service_lifecycle() -> int:
"""Test TemporalService lifecycle."""
try:
backend = InMemoryTemporalBackend()
service = TemporalService(backend=backend)
# Store initial
node = service.store_initial_node(
node_uri="uko-py:class/Bar_v1",
source_resource="RES002",
source_path="src/bar.py",
)
assert node.temporal.is_current is True
# Create revision
v2 = service.create_revision(
current_uri="uko-py:class/Bar_v1",
new_node_uri="uko-py:class/Bar_v2",
source_resource="RES002",
source_path="src/bar.py",
)
assert v2.temporal.is_revision_of == "uko-py:class/Bar_v1"
# Get current
current = service.get_current("uko-py:class/Bar")
assert current is not None
assert current.node_uri == "uko-py:class/Bar_v2"
# Get chain
chain = service.get_revision_chain("uko-py:class/Bar_v2")
assert chain.depth == 2
print("temporal-service-lifecycle-ok")
return 0
except Exception as exc:
print(f"FAIL: {exc}")
return 1
def _test_model_creation() -> int:
"""Test domain model creation."""
try:
now = datetime.now(tz=UTC)
meta = TemporalMetadata(valid_from=now)
assert meta.is_current is True
assert meta.valid_until is None
node = TemporalNode(
node_uri="uko-py:class/Test_v1",
source_resource="RES",
source_path="src/test.py",
temporal=meta,
)
assert node.node_uri == "uko-py:class/Test_v1"
chain = RevisionChain(current_uri="uko:test")
assert chain.depth == 1
result = TierQueryResult(
tier=ContextTier.HOT,
temporal_scope=TemporalScope.CURRENT,
)
assert result.nodes == ()
retention = TierRetentionConfig()
assert retention.warm_retention_hours == 24
print("temporal-model-creation-ok")
return 0
except Exception as exc:
print(f"FAIL: {exc}")
return 1
def _test_validation() -> int:
"""Test input validation on models and backend."""
try:
backend = InMemoryTemporalBackend()
# Blank URI should raise ValueError
try:
backend.get_current("")
print("FAIL: expected ValueError")
return 1
except ValueError:
pass
# Nonexistent node should raise KeyError
try:
backend.mark_historical("uko:nonexistent", datetime.now(tz=UTC))
print("FAIL: expected KeyError")
return 1
except KeyError:
pass
print("temporal-validation-ok")
return 0
except Exception as exc:
print(f"FAIL: {exc}")
return 1
def _test_retention_defaults() -> int:
"""Test retention config defaults match spec."""
try:
retention = TierRetentionConfig()
assert retention.warm_retention_hours == 24
assert retention.cold_retention_days == 90
print("temporal-retention-defaults-ok")
return 0
except Exception as exc:
print(f"FAIL: {exc}")
return 1
def _test_integration_modify_resource() -> int:
"""Integration test: modify resource, verify temporal chain.
Acceptance criteria #10: Modify a resource, verify old node is
marked historical, new node is current, revision chain is intact.
"""
try:
backend = InMemoryTemporalBackend()
service = TemporalService(backend=backend)
# Step 1: Initial analysis creates first node
v1 = service.store_initial_node(
node_uri="uko-py:class/AuthManager_v1",
source_resource="RES_AUTH",
source_path="src/auth/manager.py",
source_range="15:1-87:0",
)
assert v1.temporal.is_current is True
assert v1.temporal.is_revision_of is None
# Step 2: Code change triggers re-analysis → new revision
v2 = service.create_revision(
current_uri="uko-py:class/AuthManager_v1",
new_node_uri="uko-py:class/AuthManager_v2",
source_resource="RES_AUTH",
source_path="src/auth/manager.py",
source_range="15:1-95:0",
)
# Step 3: Verify old node is marked historical
old = backend._nodes["uko-py:class/AuthManager_v1"]
assert old.temporal.is_current is False
assert old.temporal.valid_until is not None
# Step 4: Verify new node is current
assert v2.temporal.is_current is True
assert v2.temporal.is_revision_of == "uko-py:class/AuthManager_v1"
# Step 5: Verify revision chain is intact
chain = service.get_revision_chain("uko-py:class/AuthManager_v2")
assert chain.depth == 2
assert chain.current_uri == "uko-py:class/AuthManager_v2"
assert "uko-py:class/AuthManager_v1" in chain.predecessors
# Step 6: Hot tier shows only current
hot = service.query_by_tier(ContextTier.HOT)
assert len(hot.nodes) == 1
assert hot.nodes[0].node_uri == "uko-py:class/AuthManager_v2"
# Step 7: Cold tier shows full history
cold = service.query_by_tier(ContextTier.COLD)
assert len(cold.nodes) == 2
print("temporal-integration-modify-resource-ok")
return 0
except Exception as exc:
print(f"FAIL: {exc}")
return 1
if __name__ == "__main__":
sys.exit(main())
+73
View File
@@ -0,0 +1,73 @@
*** Settings ***
Documentation Smoke tests for Temporal Data Model (Revision-Aware RDF)
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_temporal_data_model.py
*** Test Cases ***
InMemoryTemporalBackend Protocol Compliance
[Documentation] Verify InMemoryTemporalBackend satisfies TemporalBackend protocol
${result}= Run Process ${PYTHON} ${HELPER} protocol-compliance cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} temporal-protocol-ok
Revision Chain Creation And Traversal
[Documentation] Create 3-version revision chain and verify traversal
${result}= Run Process ${PYTHON} ${HELPER} revision-chain cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} temporal-revision-chain-ok
Tier Aware Temporal Queries
[Documentation] Verify hot/warm/cold tier queries with temporal filtering
${result}= Run Process ${PYTHON} ${HELPER} tier-queries cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} temporal-tier-queries-ok
TemporalService Lifecycle
[Documentation] End-to-end service lifecycle: store, revise, query, chain
${result}= Run Process ${PYTHON} ${HELPER} service-lifecycle cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} temporal-service-lifecycle-ok
Domain Model Construction
[Documentation] Verify all temporal domain models can be constructed
${result}= Run Process ${PYTHON} ${HELPER} model-creation cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} temporal-model-creation-ok
Input Validation
[Documentation] Verify backends reject invalid inputs
${result}= Run Process ${PYTHON} ${HELPER} validation cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} temporal-validation-ok
Retention Config Defaults Match Spec
[Documentation] Verify warm=24h, cold=90d defaults per spec lines 28692-28693
${result}= Run Process ${PYTHON} ${HELPER} retention-defaults cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} temporal-retention-defaults-ok
Integration Modify Resource Verify Temporal Chain
[Documentation] Modify a resource, verify old=historical, new=current, chain intact
${result}= Run Process ${PYTHON} ${HELPER} integration-modify-resource cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} temporal-integration-modify-resource-ok
@@ -163,6 +163,9 @@ from cleveragents.application.services.subplan_service import (
SpawnValidationResult,
SubplanService,
)
from cleveragents.application.services.temporal_service import (
TemporalService,
)
from cleveragents.application.services.tool_registry_service import (
ToolRegistryService,
)
@@ -298,6 +301,7 @@ __all__ = [
"SubplanService",
"SyntaxCheckRule",
"TemporalArchaeologyStrategy",
"TemporalService",
"ToolRegistryService",
"TraceService",
"UKOIndexer",
@@ -0,0 +1,365 @@
"""Temporal Data Model service for revision-aware RDF with 3 storage tiers.
Orchestrates temporal node lifecycle: creating revisions, marking nodes
as historical, querying by temporal scope and storage tier, and
traversing revision chains. All operations are delegated to an
injected :class:`TemporalBackend` implementation.
Based on ``docs/specification.md`` lines 42468--42499 (Temporal Data
Model, Revision-Aware RDF, Three Storage Tiers).
"""
from __future__ import annotations
from datetime import UTC, datetime
import structlog
from cleveragents.domain.models.acms._validation import (
validate_non_blank as _validate_non_blank,
)
from cleveragents.domain.models.acms.temporal import (
RevisionChain,
TemporalBackend,
TemporalMetadata,
TemporalNode,
TierQueryResult,
TierRetentionConfig,
)
from cleveragents.domain.models.acms.tiers import ContextTier
from cleveragents.domain.models.core.project import TemporalScope
__all__ = [
"TemporalService",
]
logger = structlog.get_logger(__name__)
class TemporalService:
"""Service layer for temporal data model operations.
Wraps a :class:`TemporalBackend` with structured logging,
input validation, and higher-level convenience methods.
Based on ``docs/specification.md`` lines 42468--42499.
Args:
backend: Injected temporal backend implementation.
retention: Tier retention policy (defaults to spec defaults).
"""
# INVARIANTS (must hold after every public method returns):
# 1. _backend is never None (set in __init__).
# 2. _retention is never None (set in __init__).
# 3. All public methods validate string inputs for non-blank.
# 4. All public methods log entry and result via structlog.
def __init__(
self,
backend: TemporalBackend,
retention: TierRetentionConfig | None = None,
) -> None:
"""Initialise the temporal service.
Args:
backend: Temporal backend for storage operations.
retention: Retention policy; defaults to spec defaults
(24h warm, 90d cold).
Raises:
TypeError: If *backend* is ``None``.
"""
if backend is None:
msg = "backend must not be None"
raise TypeError(msg)
self._backend = backend
self._retention = retention or TierRetentionConfig()
logger.info(
"temporal_service.init",
warm_hours=self._retention.warm_retention_hours,
cold_days=self._retention.cold_retention_days,
)
@property
def retention(self) -> TierRetentionConfig:
"""Current tier retention configuration."""
return self._retention
def create_revision(
self,
current_uri: str,
new_node_uri: str,
source_resource: str,
source_path: str,
source_range: str | None = None,
) -> TemporalNode:
"""Create a new revision of an existing node.
Marks the existing node (identified by *current_uri*) as
historical and creates a new head node with
``is_revision_of`` linking back.
Args:
current_uri: URI of the current (to-be-superseded) node.
new_node_uri: URI for the new revision node.
source_resource: ULID of the originating resource.
source_path: File path within the resource.
source_range: Optional source range.
Returns:
The newly created temporal node.
Raises:
ValueError: If any required string is empty or
whitespace-only, or if ``current_uri == new_node_uri``.
KeyError: If *current_uri* does not exist in the backend.
"""
_validate_non_blank(current_uri, "current_uri")
_validate_non_blank(new_node_uri, "new_node_uri")
_validate_non_blank(source_resource, "source_resource")
_validate_non_blank(source_path, "source_path")
if current_uri == new_node_uri:
msg = (
"new_node_uri must differ from current_uri; "
"same URI would overwrite the historical record"
)
raise ValueError(msg)
log = logger.bind(
current_uri=current_uri,
new_node_uri=new_node_uri,
)
log.info("temporal_service.create_revision.start")
timestamp = datetime.now(tz=UTC)
new_node = TemporalNode(
node_uri=new_node_uri,
source_resource=source_resource,
source_path=source_path,
source_range=source_range,
temporal=TemporalMetadata(
valid_from=timestamp,
is_current=True,
is_revision_of=current_uri,
),
)
result = self._backend.create_revision(
current_uri=current_uri,
new_node=new_node,
timestamp=timestamp,
)
log.info(
"temporal_service.create_revision.done",
new_uri=result.node_uri,
)
return result
def store_initial_node(
self,
node_uri: str,
source_resource: str,
source_path: str,
source_range: str | None = None,
) -> TemporalNode:
"""Store the first version of a temporal node (no predecessor).
This is used when a resource is first analysed and has no
prior revision history.
Args:
node_uri: URI for the new node.
source_resource: ULID of the originating resource.
source_path: File path within the resource.
source_range: Optional source range.
Returns:
The stored temporal node.
Raises:
ValueError: If any required string is empty or
whitespace-only.
"""
_validate_non_blank(node_uri, "node_uri")
_validate_non_blank(source_resource, "source_resource")
_validate_non_blank(source_path, "source_path")
log = logger.bind(node_uri=node_uri)
log.info("temporal_service.store_initial_node.start")
timestamp = datetime.now(tz=UTC)
node = TemporalNode(
node_uri=node_uri,
source_resource=source_resource,
source_path=source_path,
source_range=source_range,
temporal=TemporalMetadata(
valid_from=timestamp,
is_current=True,
is_revision_of=None,
),
)
self._backend.store_node(node)
log.info("temporal_service.store_initial_node.done")
return node
def get_current(self, node_uri_base: str) -> TemporalNode | None:
"""Get the current version of a node.
Hot-tier access pattern: only ``isCurrent = true`` nodes.
Args:
node_uri_base: Base URI pattern to match.
Returns:
The current node, or ``None`` if not found.
Raises:
ValueError: If *node_uri_base* is empty or whitespace-only.
"""
_validate_non_blank(node_uri_base, "node_uri_base")
log = logger.bind(node_uri_base=node_uri_base)
log.debug("temporal_service.get_current")
result = self._backend.get_current(node_uri_base)
log.debug(
"temporal_service.get_current.done",
found=result is not None,
)
return result
def get_history(
self,
node_uri_base: str,
temporal_scope: TemporalScope = TemporalScope.ALL,
) -> tuple[TemporalNode, ...]:
"""Get node versions filtered by temporal scope.
Args:
node_uri_base: Base URI pattern to match.
temporal_scope: Scope controlling which versions.
Returns:
Matching nodes ordered by ``valid_from`` descending.
Raises:
ValueError: If *node_uri_base* is empty or whitespace-only.
"""
_validate_non_blank(node_uri_base, "node_uri_base")
log = logger.bind(
node_uri_base=node_uri_base,
temporal_scope=temporal_scope.value,
)
log.debug("temporal_service.get_history")
result = self._backend.get_history(
node_uri_base, temporal_scope, self._retention
)
log.debug(
"temporal_service.get_history.done",
count=len(result),
)
return result
def get_revision_chain(self, node_uri: str) -> RevisionChain:
"""Get the full revision chain for a node.
Args:
node_uri: URI of any node in the chain.
Returns:
The complete revision chain.
Raises:
ValueError: If *node_uri* is empty or whitespace-only.
KeyError: If *node_uri* does not exist.
"""
_validate_non_blank(node_uri, "node_uri")
log = logger.bind(node_uri=node_uri)
log.debug("temporal_service.get_revision_chain")
chain = self._backend.get_revision_chain(node_uri)
log.debug(
"temporal_service.get_revision_chain.done",
depth=chain.depth,
)
return chain
def query_by_tier(
self,
tier: ContextTier,
temporal_scope: TemporalScope | None = None,
) -> TierQueryResult:
"""Query nodes aligned to a specific storage tier.
If *temporal_scope* is ``None``, uses the tier's natural
alignment from the spec:
| Tier | Default Scope |
|------|---------------|
| HOT | CURRENT |
| WARM | RECENT |
| COLD | ALL |
Args:
tier: Storage tier to query.
temporal_scope: Optional override for temporal scope.
Returns:
A :class:`TierQueryResult` with matching nodes.
"""
scope = _tier_default_scope(tier) if temporal_scope is None else temporal_scope
log = logger.bind(tier=tier.value, temporal_scope=scope.value)
log.debug("temporal_service.query_by_tier")
result = self._backend.query_by_tier(tier, scope, self._retention)
log.debug(
"temporal_service.query_by_tier.done",
node_count=len(result.nodes),
)
return result
def mark_historical(
self,
node_uri: str,
) -> TemporalNode:
"""Mark a node as historical (no longer current).
Sets ``valid_until`` to now and ``is_current`` to ``False``.
Args:
node_uri: URI of the node to mark.
Returns:
The updated node.
Raises:
ValueError: If *node_uri* is empty or whitespace-only,
or if the node is already historical.
KeyError: If *node_uri* does not exist.
"""
_validate_non_blank(node_uri, "node_uri")
log = logger.bind(node_uri=node_uri)
log.info("temporal_service.mark_historical")
timestamp = datetime.now(tz=UTC)
result = self._backend.mark_historical(node_uri, timestamp)
log.info("temporal_service.mark_historical.done")
return result
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _tier_default_scope(tier: ContextTier) -> TemporalScope:
"""Return the spec-defined default temporal scope for a tier.
Based on ``docs/specification.md`` lines 42493--42499.
"""
if tier == ContextTier.HOT:
return TemporalScope.CURRENT
if tier == ContextTier.WARM:
return TemporalScope.RECENT
return TemporalScope.ALL
@@ -38,6 +38,15 @@ Stub backends (from :mod:`~cleveragents.domain.models.acms.stubs`):
- ``InMemoryTextBackend`` -- Zero-dependency text search stub
- ``InMemoryVectorBackend`` -- Zero-dependency vector search stub
- ``InMemoryGraphBackend`` -- Zero-dependency graph query stub
- ``InMemoryTemporalBackend`` -- In-memory temporal node store stub
Temporal types (from :mod:`~cleveragents.domain.models.acms.temporal`):
- ``TemporalMetadata`` -- Temporal fields on a UKO InformationUnit
- ``TemporalNode`` -- A UKO node with temporal metadata
- ``RevisionChain`` -- Ordered revision chain (oldest-first)
- ``TierQueryResult`` -- Result from a tier-aware temporal query
- ``TierRetentionConfig`` -- Warm/cold tier retention policy
- ``TemporalBackend`` -- Protocol for temporal storage and queries
Tier types (from :mod:`~cleveragents.domain.models.acms.tiers`):
- ``ContextTier`` -- Hot/warm/cold storage tier enumeration
@@ -182,6 +191,17 @@ from cleveragents.domain.models.acms.stubs import (
InMemoryTextBackend,
InMemoryVectorBackend,
)
from cleveragents.domain.models.acms.temporal import (
RevisionChain,
TemporalBackend,
TemporalMetadata,
TemporalNode,
TierQueryResult,
TierRetentionConfig,
)
from cleveragents.domain.models.acms.temporal_stubs import (
InMemoryTemporalBackend,
)
from cleveragents.domain.models.acms.tiers import (
ActorContextView,
ActorRole,
@@ -215,6 +235,7 @@ __all__: list[str] = [
"GraphResult",
"InMemoryGraphBackend",
"InMemoryGraphIndexBackend",
"InMemoryTemporalBackend",
"InMemoryTextBackend",
"InMemoryTextIndexBackend",
"InMemoryVectorBackend",
@@ -230,6 +251,7 @@ __all__: list[str] = [
"PythonAnalyzer",
"ResourceAliasResolver",
"ResourceScope",
"RevisionChain",
"ScopeViolationError",
"ScopedBackendSet",
"ScopedBackendView",
@@ -240,11 +262,16 @@ __all__: list[str] = [
"StrategyConfig",
"StrategyRegistryEntry",
"TemporalArchaeologyStrategy",
"TemporalBackend",
"TemporalMetadata",
"TemporalNode",
"TextBackend",
"TextIndexBackend",
"TextResult",
"TierBudget",
"TierMetrics",
"TierQueryResult",
"TierRetentionConfig",
"TieredFragment",
"TurtleValidationError",
"UKOTriple",
@@ -0,0 +1,41 @@
"""Shared validation helpers for the ACMS domain models."""
from __future__ import annotations
import unicodedata
__all__ = ["validate_non_blank"]
# Zero-width and invisible format characters that ``str.strip()``
# does not remove (Unicode category Cf).
_INVISIBLE_CODEPOINTS = frozenset(
"\u200b" # ZERO WIDTH SPACE
"\u200c" # ZERO WIDTH NON-JOINER
"\u200d" # ZERO WIDTH JOINER
"\ufeff" # BYTE ORDER MARK / ZERO WIDTH NO-BREAK SPACE
)
def _is_visible(ch: str) -> bool:
"""Return ``True`` if *ch* is a visible (non-whitespace, non-format) character."""
if ch in _INVISIBLE_CODEPOINTS:
return False
cat = unicodedata.category(ch)
# Zs = space separator, Cf = format character, Cc = control
return cat not in {"Zs", "Cf", "Cc"}
def validate_non_blank(value: str, field_name: str) -> str:
"""Raise ``ValueError`` if *value* is empty, whitespace-only, or invisible.
Also rejects strings composed entirely of zero-width Unicode
characters (U+200B, U+200C, U+200D, U+FEFF) that pass
``str.strip()`` / ``str.isspace()`` checks.
Returns *value* unchanged so the function can be used as a
pass-through in Pydantic ``@field_validator`` methods.
"""
if not value or not any(_is_visible(ch) for ch in value):
msg = f"{field_name} must be a non-empty, non-whitespace string"
raise ValueError(msg)
return value
@@ -27,6 +27,7 @@ from cleveragents.domain.models.acms.crp import (
ContextFragment,
ContextRequest,
)
from cleveragents.domain.models.acms.temporal import TemporalBackend
__all__ = [
"BackendSet",
@@ -38,6 +39,29 @@ __all__ = [
"StrategyRegistryEntry",
]
def _deep_freeze_value(v: Any) -> Any:
"""Recursively freeze a value: dicts → MappingProxyType, lists → tuples."""
if isinstance(v, (dict, MappingProxyType)):
source = dict(v) if isinstance(v, MappingProxyType) else v
return MappingProxyType(
Outdated
Review

P2:should-fix_deep_freeze_value does not recurse into tuple contents.

Dicts, lists, and sets are recursively frozen, but tuples pass through as-is. If extra contains a tuple-of-dicts (e.g., {"key": ({"nested": "dict"},)}), the inner dict remains mutable, violating the ADR-004 deep-freeze intent.

Suggested fix — add a tuple branch:

def _deep_freeze_value(v: Any) -> Any:
    if isinstance(v, (dict, MappingProxyType)):
        source = dict(v) if isinstance(v, MappingProxyType) else v
        return MappingProxyType(
            {k: _deep_freeze_value(val) for k, val in source.items()}
        )
    if isinstance(v, list):
        return tuple(_deep_freeze_value(item) for item in v)
    if isinstance(v, tuple):
        return tuple(_deep_freeze_value(item) for item in v)
    if isinstance(v, set):
        return frozenset(_deep_freeze_value(item) for item in v)
    return v
**P2:should-fix** — `_deep_freeze_value` does not recurse into tuple contents. Dicts, lists, and sets are recursively frozen, but tuples pass through as-is. If `extra` contains a tuple-of-dicts (e.g., `{"key": ({"nested": "dict"},)}`), the inner dict remains mutable, violating the ADR-004 deep-freeze intent. Suggested fix — add a tuple branch: ```python def _deep_freeze_value(v: Any) -> Any: if isinstance(v, (dict, MappingProxyType)): source = dict(v) if isinstance(v, MappingProxyType) else v return MappingProxyType( {k: _deep_freeze_value(val) for k, val in source.items()} ) if isinstance(v, list): return tuple(_deep_freeze_value(item) for item in v) if isinstance(v, tuple): return tuple(_deep_freeze_value(item) for item in v) if isinstance(v, set): return frozenset(_deep_freeze_value(item) for item in v) return v ```
{k: _deep_freeze_value(val) for k, val in source.items()}
)
if isinstance(v, list):
return tuple(_deep_freeze_value(item) for item in v)
if isinstance(v, set):
return frozenset(_deep_freeze_value(item) for item in v)
return v
def _deep_freeze_mapping(m: Any) -> MappingProxyType[str, Any]:
"""Recursively freeze a mapping and all nested containers."""
source = dict(m) if isinstance(m, MappingProxyType) else m
return MappingProxyType(
{k: _deep_freeze_value(v) for k, v in source.items()},
)
# ---------------------------------------------------------------------------
# BackendSet — container for available data backends
# ---------------------------------------------------------------------------
@@ -65,12 +89,12 @@ class BackendSet(BaseModel, frozen=True):
default=None,
description="Knowledge graph backend (e.g., Blazegraph, Neo4j)",
)
temporal: object | None = Field(
temporal: TemporalBackend | None = Field(
default=None,
description=(
"Temporal/cold-tier backend for historical pattern discovery. "
"Typed as ``object`` until a formal TemporalBackend protocol "
"is defined; presence (not ``None``) indicates availability."
"Temporal backend for historical pattern discovery and "
"revision-aware RDF queries. Presence (not ``None``) "
"indicates temporal/cold-tier data is available."
),
)
@@ -275,9 +299,9 @@ class StrategyConfig(BaseModel, frozen=True):
) -> MappingProxyType[str, Any]:
"""Wrap plain dicts in MappingProxyType for immutability (ADR-004)."""
if isinstance(v, MappingProxyType):
return v
return _deep_freeze_mapping(v)
if isinstance(v, dict):
return MappingProxyType(v)
return _deep_freeze_mapping(v)
msg = f"expected dict or MappingProxyType, got {type(v).__name__}"
raise ValueError(msg)
@@ -351,9 +375,9 @@ class ContextStrategyResult(BaseModel, frozen=True):
) -> MappingProxyType[str, Any]:
"""Wrap plain dicts in MappingProxyType for immutability (ADR-004)."""
if isinstance(v, MappingProxyType):
return v
return _deep_freeze_mapping(v)
if isinstance(v, dict):
return MappingProxyType(v)
return _deep_freeze_mapping(v)
msg = f"expected dict or MappingProxyType, got {type(v).__name__}"
raise ValueError(msg)
@@ -0,0 +1,486 @@
"""Temporal Data Model (Revision-Aware RDF) with 3 storage tiers.
Provides the temporal metadata, revision chain, and tier-aware query
models for the ACMS. When code changes, existing UKO nodes are **not
deleted** -- they are marked historical (``valid_until`` set,
``is_current`` set to ``False``) and a new revision node is created
with ``is_revision_of`` linking back to the predecessor.
Domain models (all frozen / immutable):
| Type | Role |
|------------------------|-----------------------------------------------|
| ``TemporalMetadata`` | Temporal fields on a UKO InformationUnit |
| ``TemporalNode`` | A UKO node carrying temporal metadata |
| ``RevisionChain`` | Ordered chain of node versions (oldest-first) |
| ``TierQueryResult`` | Result from a tier-aware temporal query |
| ``TierRetentionConfig``| Warm/cold tier retention policy |
Protocol:
| Protocol | Role |
|----------------------|-------------------------------------------------|
| ``TemporalBackend`` | Interface for temporal storage and queries |
Based on ``docs/specification.md`` lines 41940--42499 (UKO temporal
properties, Revision-Aware RDF, Three Storage Tiers).
"""
from __future__ import annotations
from datetime import UTC, datetime
from typing import Protocol, runtime_checkable
from pydantic import BaseModel, Field, field_validator, model_validator
from cleveragents.domain.models.acms._validation import (
validate_non_blank as _validate_non_blank,
)
from cleveragents.domain.models.acms.tiers import ContextTier
from cleveragents.domain.models.core.project import TemporalScope
__all__ = [
"RevisionChain",
"TemporalBackend",
"TemporalMetadata",
"TemporalNode",
"TierQueryResult",
"TierRetentionConfig",
]
# ---------------------------------------------------------------------------
# TemporalMetadata — temporal fields on a UKO InformationUnit
# ---------------------------------------------------------------------------
class TemporalMetadata(BaseModel, frozen=True):
"""Temporal metadata carried by every UKO InformationUnit.
Based on ``docs/specification.md`` lines 41940--41957.
Attributes:
valid_from: UTC timestamp when this node version became valid.
Maps to ``uko:validFrom`` (``xsd:dateTime``).
valid_until: UTC timestamp when this node version was superseded.
``None`` for current (active) nodes.
Maps to ``uko:validUntil`` (``xsd:dateTime``).
is_current: Whether this is the current/active version.
Maps to ``uko:isCurrent`` (``xsd:boolean``).
is_revision_of: URI of the predecessor node, or ``None`` for the
first version in a revision chain.
Maps to ``uko:isRevisionOf`` (``owl:ObjectProperty``).
"""
valid_from: datetime = Field(
...,
description="UTC timestamp when this node version became valid",
)
valid_until: datetime | None = Field(
default=None,
description="UTC timestamp when superseded; None if current",
)
is_current: bool = Field(
default=True,
description="Whether this is the current/active version",
)
is_revision_of: str | None = Field(
default=None,
description="URI of predecessor node; None for first version",
)
@field_validator("valid_from", mode="after")
@classmethod
def _ensure_valid_from_tz(
cls: type[TemporalMetadata],
v: datetime,
) -> datetime:
"""Coerce ``valid_from`` to UTC (naive → assume UTC, aware → convert)."""
if v.tzinfo is None:
return v.replace(tzinfo=UTC)
return v.astimezone(UTC)
@field_validator("valid_until", mode="after")
@classmethod
def _ensure_valid_until_tz(
cls: type[TemporalMetadata],
v: datetime | None,
) -> datetime | None:
"""Coerce ``valid_until`` to UTC when set."""
if v is None:
return None
if v.tzinfo is None:
return v.replace(tzinfo=UTC)
return v.astimezone(UTC)
@field_validator("is_revision_of")
Outdated
Review

P3:nit — This uses inline not v or not v.strip() instead of the shared _validate_non_blank(v, "is_revision_of") helper that the TemporalNode validators use. Minor inconsistency; using the shared helper would reduce the chance of divergent validation logic.

**P3:nit** — This uses inline `not v or not v.strip()` instead of the shared `_validate_non_blank(v, "is_revision_of")` helper that the `TemporalNode` validators use. Minor inconsistency; using the shared helper would reduce the chance of divergent validation logic.
@classmethod
def _validate_revision_ref(
cls: type[TemporalMetadata],
v: str | None,
) -> str | None:
"""Reject empty, whitespace-only, or invisible revision references."""
if v is not None:
_validate_non_blank(v, "is_revision_of")
return v
@model_validator(mode="after")
def _validate_temporal_invariants(self) -> TemporalMetadata:
"""Enforce cross-field temporal invariants."""
if self.valid_until is not None and self.valid_until < self.valid_from:
msg = (
"valid_until must not be earlier than valid_from "
f"({self.valid_until} < {self.valid_from})"
)
raise ValueError(msg)
if self.is_current and self.valid_until is not None:
msg = (
"is_current=True contradicts valid_until being set; "
"a current node cannot have an expiry timestamp"
)
raise ValueError(msg)
if not self.is_current and self.valid_until is None:
msg = (
"is_current=False requires valid_until to be set; "
"a historical node must record when it was superseded"
)
raise ValueError(msg)
return self
# ---------------------------------------------------------------------------
# TemporalNode — a UKO node with temporal metadata
# ---------------------------------------------------------------------------
class TemporalNode(BaseModel, frozen=True):
"""A UKO InformationUnit carrying full temporal metadata.
Represents a single versioned node in the knowledge graph. Each
node has provenance back to the originating source resource and
carries a :class:`TemporalMetadata` for revision tracking.
Based on ``docs/specification.md`` lines 42454--42491.
Attributes:
node_uri: Unique UKO URI for this node version
(e.g., ``uko-py:class/AuthManager_v2``).
source_resource: ULID of the originating resource.
source_path: File path within the resource.
source_range: Optional source range (e.g., ``"15:1-87:0"``).
temporal: Embedded temporal metadata.
"""
node_uri: str = Field(
...,
min_length=1,
description="Unique UKO URI for this node version",
)
source_resource: str = Field(
...,
min_length=1,
description="ULID of the originating resource",
)
source_path: str = Field(
...,
min_length=1,
description="File path within the resource",
)
source_range: str | None = Field(
default=None,
description='Optional source range (e.g., "15:1-87:0")',
)
temporal: TemporalMetadata = Field(
...,
description="Temporal metadata for revision tracking",
)
@field_validator("node_uri")
@classmethod
def _validate_node_uri(
cls: type[TemporalNode],
v: str,
) -> str:
"""Reject whitespace-only node URIs."""
return _validate_non_blank(v, "node_uri")
@field_validator("source_resource")
@classmethod
def _validate_source_resource(
cls: type[TemporalNode],
v: str,
) -> str:
"""Reject whitespace-only source resource IDs."""
return _validate_non_blank(v, "source_resource")
@field_validator("source_path")
@classmethod
def _validate_source_path(
cls: type[TemporalNode],
v: str,
) -> str:
"""Reject whitespace-only source paths."""
return _validate_non_blank(v, "source_path")
@field_validator("source_range")
@classmethod
def _validate_source_range(
cls: type[TemporalNode],
v: str | None,
) -> str | None:
"""Reject empty or whitespace-only source ranges."""
if v is not None and (not v or not v.strip()):
msg = "source_range must be a non-empty, non-whitespace string or None"
raise ValueError(msg)
return v
# ---------------------------------------------------------------------------
# RevisionChain — ordered chain of node versions
# ---------------------------------------------------------------------------
class RevisionChain(BaseModel, frozen=True):
"""Ordered chain of temporal node versions for a single UKO concept.
The chain captures the full revision history: ``predecessors`` are
ordered oldest-first, and ``current_uri`` identifies the active
(``isCurrent = true``) head of the chain.
Based on ``docs/specification.md`` lines 42468--42491.
Attributes:
current_uri: URI of the current (active) head node.
predecessors: URIs of predecessor nodes, oldest-first.
"""
current_uri: str = Field(
...,
min_length=1,
description="URI of the current (active) head node",
)
predecessors: tuple[str, ...] = Field(
default=(),
description="URIs of predecessor nodes, oldest-first",
)
@field_validator("current_uri")
@classmethod
def _validate_current_uri(
cls: type[RevisionChain],
v: str,
) -> str:
"""Reject whitespace-only current URIs."""
return _validate_non_blank(v, "current_uri")
@field_validator("predecessors")
@classmethod
def _validate_predecessors(
cls: type[RevisionChain],
v: tuple[str, ...],
) -> tuple[str, ...]:
"""Reject empty, whitespace-only, or invisible predecessor URIs."""
for i, uri in enumerate(v):
_validate_non_blank(uri, f"predecessors[{i}]")
return v
Outdated
Review

P2:should-fixRevisionChain accepts duplicate URIs in predecessors.

The _validate_no_duplicate_current model validator checks that current_uri is not in predecessors, but there is no check that predecessors itself contains unique URIs. A chain like RevisionChain(current_uri="c", predecessors=("a", "a")) would be silently accepted, representing logically invalid data.

Suggested fix:

@model_validator(mode="after")
def _validate_no_duplicate_current(self) -> RevisionChain:
    if self.current_uri in self.predecessors:
        msg = f"current_uri {self.current_uri!r} must not appear in predecessors"
        raise ValueError(msg)
    if len(self.predecessors) != len(set(self.predecessors)):
        msg = "predecessors must not contain duplicate URIs"
        raise ValueError(msg)
    return self
**P2:should-fix** — `RevisionChain` accepts duplicate URIs in `predecessors`. The `_validate_no_duplicate_current` model validator checks that `current_uri` is not in `predecessors`, but there is no check that `predecessors` itself contains unique URIs. A chain like `RevisionChain(current_uri="c", predecessors=("a", "a"))` would be silently accepted, representing logically invalid data. Suggested fix: ```python @model_validator(mode="after") def _validate_no_duplicate_current(self) -> RevisionChain: if self.current_uri in self.predecessors: msg = f"current_uri {self.current_uri!r} must not appear in predecessors" raise ValueError(msg) if len(self.predecessors) != len(set(self.predecessors)): msg = "predecessors must not contain duplicate URIs" raise ValueError(msg) return self ```
@model_validator(mode="after")
def _validate_chain_integrity(self) -> RevisionChain:
"""Ensure chain URIs are unique and current is not a predecessor."""
if self.current_uri in self.predecessors:
msg = f"current_uri {self.current_uri!r} must not appear in predecessors"
raise ValueError(msg)
if len(self.predecessors) != len(set(self.predecessors)):
msg = "predecessors must not contain duplicate URIs"
raise ValueError(msg)
return self
@property
def depth(self) -> int:
"""Number of revisions in the chain (predecessors + current)."""
return len(self.predecessors) + 1
@property
def all_uris(self) -> tuple[str, ...]:
"""All URIs in order: predecessors (oldest-first) then current."""
return (*self.predecessors, self.current_uri)
# ---------------------------------------------------------------------------
# TierRetentionConfig — warm/cold tier retention policy
# ---------------------------------------------------------------------------
class TierRetentionConfig(BaseModel, frozen=True):
"""Retention policy for warm and cold storage tiers.
Based on ``docs/specification.md`` lines 28692--28693.
Attributes:
warm_retention_hours: Hours warm-tier entries are retained
before demotion to cold. Maps to config key
``context.tiers.warm.retention-hours``
(env: ``CLEVERAGENTS_CTX_WARM_HOURS``).
cold_retention_days: Days cold-tier entries are retained
before archival/expiry. Maps to config key
``context.tiers.cold.retention-days``
(env: ``CLEVERAGENTS_CTX_COLD_DAYS``).
"""
warm_retention_hours: int = Field(
default=24,
ge=1,
description="Hours warm-tier entries are retained (default 24h)",
)
cold_retention_days: int = Field(
default=90,
ge=1,
description="Days cold-tier entries are retained (default 90d)",
)
# ---------------------------------------------------------------------------
# TierQueryResult — result from a tier-aware temporal query
# ---------------------------------------------------------------------------
class TierQueryResult(BaseModel, frozen=True):
"""Result from a tier-aware temporal query.
Based on ``docs/specification.md`` lines 42493--42499 (Three
Storage Tiers with Temporal Alignment).
Attributes:
nodes: Matching temporal nodes.
tier: Which storage tier was queried.
temporal_scope: Temporal scope applied to the query.
"""
nodes: tuple[TemporalNode, ...] = Field(
default=(),
description="Matching temporal nodes",
)
tier: ContextTier = Field(
...,
description="Storage tier that was queried (HOT/WARM/COLD)",
)
temporal_scope: TemporalScope = Field(
...,
description="Temporal scope applied (CURRENT/RECENT/ALL)",
)
# ---------------------------------------------------------------------------
# TemporalBackend — protocol for temporal storage and queries
# ---------------------------------------------------------------------------
@runtime_checkable
class TemporalBackend(Protocol):
"""Protocol for temporal storage and revision-aware queries.
Replaces the ``BackendSet.temporal: object | None`` placeholder
with a formal typed interface. Based on spec lines 42468--42591.
Lifecycle::
backend = InMemoryTemporalBackend()
node = backend.create_revision(
"uko-py:class/Foo_v1", new_node, datetime.now(tz=UTC))
chain = backend.get_revision_chain("uko-py:class/Foo_v2")
"""
def create_revision(
self,
current_uri: str,
new_node: TemporalNode,
timestamp: datetime,
) -> TemporalNode:
"""Create a new revision, marking *current_uri* historical.
Raises:
ValueError: If *current_uri* is blank, equals
``new_node.node_uri``, the node is already historical,
``new_node.node_uri`` already exists,
``new_node.temporal.is_revision_of != current_uri``,
or ``new_node.temporal.is_current`` is not ``True``.
KeyError: If *current_uri* does not exist.
"""
...
def get_current(
self,
node_uri_base: str,
) -> TemporalNode | None:
"""Return the current (``isCurrent = true``) version.
Raises:
ValueError: If *node_uri_base* is empty or whitespace-only.
"""
...
def get_history(
self,
node_uri_base: str,
temporal_scope: TemporalScope,
retention: TierRetentionConfig | None = None,
) -> tuple[TemporalNode, ...]:
"""Return versions filtered by scope (CURRENT/RECENT/ALL).
Args:
node_uri_base: Base URI pattern to match.
temporal_scope: Scope controlling which versions to include.
retention: Optional retention config for RECENT scope.
Defaults to :class:`TierRetentionConfig` defaults when
``None``.
Raises:
ValueError: If *node_uri_base* is empty or whitespace-only.
"""
...
def get_revision_chain(
self,
node_uri: str,
) -> RevisionChain:
"""Traverse the full revision chain for *node_uri*.
Raises:
ValueError: If *node_uri* is empty or whitespace-only.
KeyError: If *node_uri* does not exist.
"""
...
def query_by_tier(
self,
tier: ContextTier,
temporal_scope: TemporalScope,
retention: TierRetentionConfig,
) -> TierQueryResult:
"""Query nodes aligned to a storage tier with temporal scope."""
...
def store_node(self, node: TemporalNode) -> None:
"""Store a temporal node directly (initial insert).
Used by :meth:`TemporalService.store_initial_node` to persist
the first version of a node that has no predecessor.
Raises:
ValueError: If a node with the same ``node_uri`` already
exists.
"""
...
def mark_historical(
self,
node_uri: str,
timestamp: datetime,
) -> TemporalNode:
"""Mark a node historical (``valid_until`` set, ``is_current`` cleared).
Raises:
ValueError: If *node_uri* is empty or whitespace-only,
or if the node is already historical.
KeyError: If *node_uri* does not exist.
"""
...
@@ -0,0 +1,482 @@
"""In-memory stub backend for the ACMS Temporal Data Model.
Provides :class:`InMemoryTemporalBackend`, a zero-dependency
implementation of the :class:`TemporalBackend` protocol that stores
:class:`TemporalNode` instances in a ``dict`` keyed by ``node_uri``.
Unlike the BAL stubs in ``stubs.py`` (which return empty results),
this backend stores real data so revision chain creation and temporal
queries can be exercised in tests without external infrastructure.
Based on ``docs/specification.md`` lines 42468--42499 (Temporal Data
Model / Revision-Aware RDF / Three Storage Tiers).
"""
from __future__ import annotations
import warnings
from datetime import UTC, datetime, timedelta
from cleveragents.domain.models.acms._validation import (
validate_non_blank as _validate_non_blank,
)
from cleveragents.domain.models.acms.temporal import (
RevisionChain,
TemporalMetadata,
TemporalNode,
TierQueryResult,
TierRetentionConfig,
)
from cleveragents.domain.models.acms.tiers import ContextTier
from cleveragents.domain.models.core.project import TemporalScope
__all__ = [
"InMemoryTemporalBackend",
]
def _uri_matches_base(node_uri: str, base: str) -> bool:
"""Check whether *node_uri* belongs to the URI family of *base*.
Returns ``True`` when *node_uri* starts with *base* **and** the
character immediately after the base prefix is either ``_`` (the
version delimiter) or the string ends exactly at *base*. This
prevents ``"uko-py:class/Auth"`` from matching
``"uko-py:class/AuthManager_v1"``.
"""
if not node_uri.startswith(base):
return False
# Exact match or next char is the version delimiter ``_``
return len(node_uri) == len(base) or node_uri[len(base)] == "_"
class InMemoryTemporalBackend:
"""Stub :class:`TemporalBackend` with in-memory node storage.
Stores :class:`TemporalNode` instances in a ``dict`` keyed by
``node_uri`` so revision chain creation and temporal queries can
be exercised in tests without external infrastructure.
Based on ``docs/specification.md`` lines 42468--42499.
"""
# INVARIANTS (must hold after every public method returns):
# 1. Every key in _nodes is the node_uri of its corresponding value.
# 2. At most one node with a given base URI has is_current == True.
# 3. _nodes values are TemporalNode (frozen); internal state never
# exposes mutable references.
def __init__(self) -> None:
"""Initialise with an empty node store."""
self._nodes: dict[str, TemporalNode] = {}
# Reverse index: predecessor_uri → list of successor URIs.
# Maintained by store_node / create_revision for O(k) chain walks.
self._successors: dict[str, list[str]] = {}
def create_revision(
self,
current_uri: str,
new_node: TemporalNode,
timestamp: datetime,
) -> TemporalNode:
"""Create a new revision, marking *current_uri* as historical.
Args:
current_uri: URI of the existing current node.
new_node: New temporal node to store.
timestamp: UTC timestamp of the revision.
Returns:
The stored new node.
Raises:
ValueError: If *current_uri* is empty or whitespace-only,
if ``new_node.node_uri == current_uri``, if the
node at *current_uri* is already historical, if
``new_node.node_uri`` already exists in the store,
if ``new_node.temporal.is_revision_of != current_uri``,
or if ``new_node.temporal.is_current`` is not ``True``.
KeyError: If *current_uri* does not exist.
"""
_validate_non_blank(current_uri, "current_uri")
if current_uri not in self._nodes:
raise KeyError(current_uri)
if new_node.node_uri == current_uri:
msg = (
"new_node.node_uri must differ from current_uri; "
"same URI would overwrite the historical record"
)
raise ValueError(msg)
if new_node.node_uri in self._nodes:
msg = (
f"new_node.node_uri {new_node.node_uri!r} already exists; "
"create_revision would overwrite an unrelated node"
)
raise ValueError(msg)
if new_node.temporal.is_revision_of != current_uri:
msg = (
f"new_node.temporal.is_revision_of must be {current_uri!r}, "
f"got {new_node.temporal.is_revision_of!r}"
)
raise ValueError(msg)
if not new_node.temporal.is_current:
msg = "new_node.temporal.is_current must be True for a new revision"
raise ValueError(msg)
old_node = self._nodes[current_uri]
if not old_node.temporal.is_current:
msg = (
f"cannot revise non-current node {current_uri!r}; "
"node is already historical"
)
raise ValueError(msg)
# Mark old node as historical
historical_meta = TemporalMetadata(
valid_from=old_node.temporal.valid_from,
valid_until=timestamp,
is_current=False,
is_revision_of=old_node.temporal.is_revision_of,
)
historical_node = TemporalNode(
node_uri=old_node.node_uri,
source_resource=old_node.source_resource,
source_path=old_node.source_path,
source_range=old_node.source_range,
temporal=historical_meta,
)
# Update tracking BEFORE storing (lesson learned from PR #612)
self._nodes[current_uri] = historical_node
# Store new node and update successor index.
self._nodes[new_node.node_uri] = new_node
pred = new_node.temporal.is_revision_of
if pred is not None:
self._successors.setdefault(pred, []).append(
new_node.node_uri,
)
return new_node
def get_current(
self,
node_uri_base: str,
) -> TemporalNode | None:
"""Return the current version matching *node_uri_base*.
Note:
URI matching is delimiter-aware: the base must be followed
by ``_`` or match exactly, preventing prefix collisions
(e.g., ``"Auth"`` will not match ``"AuthManager_v1"``).
Raises:
ValueError: If *node_uri_base* is empty or whitespace-only.
"""
_validate_non_blank(node_uri_base, "node_uri_base")
for node in self._nodes.values():
if (
_uri_matches_base(node.node_uri, node_uri_base)
and node.temporal.is_current
):
return node
return None
def get_history(
self,
node_uri_base: str,
temporal_scope: TemporalScope,
retention: TierRetentionConfig | None = None,
) -> tuple[TemporalNode, ...]:
"""Return node versions filtered by temporal scope.
Args:
node_uri_base: Base URI pattern to match.
temporal_scope: Scope controlling which versions to include.
retention: Optional retention config for RECENT scope.
Defaults to :class:`TierRetentionConfig` defaults when
``None`` (24 h warm, 90 d cold).
Returns:
Matching nodes ordered by ``valid_from`` descending.
Note:
URI matching is delimiter-aware: the base must be followed
by ``_`` or match exactly, preventing prefix collisions.
Raises:
ValueError: If *node_uri_base* is empty or whitespace-only.
"""
_validate_non_blank(node_uri_base, "node_uri_base")
matching = [
n
for n in self._nodes.values()
if _uri_matches_base(n.node_uri, node_uri_base)
]
if temporal_scope == TemporalScope.CURRENT:
matching = [n for n in matching if n.temporal.is_current]
elif temporal_scope == TemporalScope.RECENT:
effective = retention or TierRetentionConfig()
cutoff = datetime.now(tz=UTC) - timedelta(
hours=effective.warm_retention_hours,
)
matching = [
n
for n in matching
if n.temporal.is_current
or (
n.temporal.valid_until is not None
and n.temporal.valid_until >= cutoff
)
]
# TemporalScope.ALL returns everything
return tuple(
sorted(
matching,
key=lambda n: n.temporal.valid_from,
reverse=True,
),
)
def get_revision_chain(
self,
node_uri: str,
) -> RevisionChain:
"""Traverse the full revision chain for *node_uri*.
Args:
node_uri: URI of any node in the chain.
Returns:
The complete revision chain.
Note:
Uses backward-walk from *node_uri* to the root, then a
single-successor forward-walk via the ``_successors``
index to the current head. This avoids branch
contamination: only the linear path through the queried
node is returned. ``O(k)`` where *k* is chain length.
Raises:
ValueError: If *node_uri* is empty or whitespace-only,
or if no current node exists in the chain (all
historical).
KeyError: If *node_uri* does not exist.
"""
_validate_non_blank(node_uri, "node_uri")
if node_uri not in self._nodes:
raise KeyError(node_uri)
# Walk backwards from the given node to the root.
backward: list[str] = [node_uri]
backward_set: set[str] = {node_uri}
walker = self._nodes[node_uri]
while walker.temporal.is_revision_of is not None:
pred_uri = walker.temporal.is_revision_of
if pred_uri in backward_set:
break # cycle guard
if pred_uri not in self._nodes:
warnings.warn(
f"predecessor {pred_uri!r} not found in store; "
f"chain for {node_uri!r} may be incomplete",
stacklevel=2,
)
break
backward.append(pred_uri)
backward_set.add(pred_uri)
walker = self._nodes[pred_uri]
Outdated
Review

P3:nit — When pred_uri not in self._nodes (line 291), the backward walk silently stops. This could mask data integrity issues in tests where a predecessor reference points to a node that was never stored. Consider emitting a structlog.warning here so broken chains are discoverable during test runs.

**P3:nit** — When `pred_uri not in self._nodes` (line 291), the backward walk silently stops. This could mask data integrity issues in tests where a predecessor reference points to a node that was never stored. Consider emitting a `structlog.warning` here so broken chains are discoverable during test runs.
# Reverse to get root → queried_node order.
chain_uris = list(reversed(backward))
# Walk forward from queried_node to current head using the
# _successors index (O(k) instead of O(n*k) full scan).
tip = node_uri
forward_seen: set[str] = set(chain_uris)
while True:
children = self._successors.get(tip, [])
# Pick successor by (latest valid_from, lexicographic URI)
# for fully deterministic branch choice regardless of
# insertion order.
successor: str | None = None
best_key: tuple[datetime, str] | None = None
for child_uri in children:
if child_uri in forward_seen:
continue
vf = self._nodes[child_uri].temporal.valid_from
key = (vf, child_uri)
if best_key is None or key > best_key:
successor = child_uri
best_key = key
if successor is None:
break
Outdated
Review

P3:nit — Comment says "deterministic branch choice" but the tie-breaking when two successors have identical valid_from timestamps depends on their insertion order in _successors (the last appended wins since > is strict). This is stable for a given insertion sequence but not truly deterministic in the abstract sense — consider >= with a secondary tiebreaker (e.g., lexicographic URI) or documenting the insertion-order dependency.

**P3:nit** — Comment says "deterministic branch choice" but the tie-breaking when two successors have identical `valid_from` timestamps depends on their insertion order in `_successors` (the last appended wins since `>` is strict). This is stable for a given insertion sequence but not truly deterministic in the abstract sense — consider `>=` with a secondary tiebreaker (e.g., lexicographic URI) or documenting the insertion-order dependency.
chain_uris.append(successor)
forward_seen.add(successor)
tip = successor
# Find the current head explicitly.
head_uri: str | None = None
for uri in chain_uris:
if self._nodes[uri].temporal.is_current:
head_uri = uri
break
if head_uri is None:
msg = (
f"no current node in revision chain for {node_uri!r}; "
"all nodes are historical"
)
raise ValueError(msg)
predecessors_tuple = tuple(u for u in chain_uris if u != head_uri)
return RevisionChain(
current_uri=head_uri,
predecessors=predecessors_tuple,
)
def query_by_tier(
self,
tier: ContextTier,
temporal_scope: TemporalScope,
retention: TierRetentionConfig,
) -> TierQueryResult:
"""Query nodes aligned to a specific storage tier.
Args:
tier: Storage tier to query.
temporal_scope: Temporal scope for filtering.
retention: Retention configuration.
Returns:
A :class:`TierQueryResult` with matching nodes.
"""
now = datetime.now(tz=UTC)
all_nodes = list(self._nodes.values())
# Step 1: Tier determines the data pool (retention window).
if tier == ContextTier.HOT:
pool = [n for n in all_nodes if n.temporal.is_current]
elif tier == ContextTier.WARM:
cutoff = now - timedelta(hours=retention.warm_retention_hours)
pool = [
n
for n in all_nodes
if n.temporal.is_current
or (
n.temporal.valid_until is not None
and n.temporal.valid_until >= cutoff
)
]
else:
# Cold: all temporal versions within cold retention
cutoff = now - timedelta(days=retention.cold_retention_days)
pool = [
n
for n in all_nodes
if n.temporal.is_current
or (
n.temporal.valid_until is not None
and n.temporal.valid_until >= cutoff
)
]
# Step 2: temporal_scope filters within the tier pool.
if temporal_scope == TemporalScope.CURRENT:
filtered = [n for n in pool if n.temporal.is_current]
elif temporal_scope == TemporalScope.RECENT:
scope_cutoff = now - timedelta(
hours=retention.warm_retention_hours,
)
filtered = [
n
for n in pool
if n.temporal.is_current
or (
n.temporal.valid_until is not None
and n.temporal.valid_until >= scope_cutoff
)
]
else:
# TemporalScope.ALL — no additional filtering
filtered = pool
nodes_tuple = tuple(
sorted(
filtered,
key=lambda n: n.temporal.valid_from,
reverse=True,
),
)
return TierQueryResult(
nodes=nodes_tuple,
tier=tier,
temporal_scope=temporal_scope,
)
def mark_historical(
self,
node_uri: str,
timestamp: datetime,
) -> TemporalNode:
"""Mark a node as historical.
Args:
node_uri: URI of the node to mark.
timestamp: UTC timestamp when superseded.
Returns:
The updated node.
Raises:
ValueError: If *node_uri* is empty or whitespace-only,
or if the node is already historical.
KeyError: If *node_uri* does not exist.
"""
_validate_non_blank(node_uri, "node_uri")
if node_uri not in self._nodes:
raise KeyError(node_uri)
old_node = self._nodes[node_uri]
if not old_node.temporal.is_current:
msg = (
f"node {node_uri!r} is already historical "
f"(valid_until={old_node.temporal.valid_until})"
)
raise ValueError(msg)
historical_meta = TemporalMetadata(
valid_from=old_node.temporal.valid_from,
valid_until=timestamp,
is_current=False,
is_revision_of=old_node.temporal.is_revision_of,
)
updated = TemporalNode(
node_uri=old_node.node_uri,
source_resource=old_node.source_resource,
source_path=old_node.source_path,
source_range=old_node.source_range,
temporal=historical_meta,
)
self._nodes[node_uri] = updated
return updated
def store_node(self, node: TemporalNode) -> None:
"""Store a temporal node directly (initial insert).
Args:
node: The temporal node to store.
Raises:
ValueError: If a node with the same ``node_uri`` already
exists.
"""
if node.node_uri in self._nodes:
msg = f"node {node.node_uri!r} already exists"
raise ValueError(msg)
self._nodes[node.node_uri] = node
pred = node.temporal.is_revision_of
if pred is not None:
self._successors.setdefault(pred, []).append(node.node_uri)