Add configurable ecosystem coordination frame with one-hour baseline
All checks were successful
check / test (push) Successful in 1m23s
CI Smoke / container-smoke (push) Successful in 2s
CI Smoke / host-smoke (push) Successful in 0s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
This commit is contained in:
tegwick 2026-10-15 20:41:33 +02:00
parent 16de8fc3f7
commit f658556690
4 changed files with 173 additions and 0 deletions

View file

@ -0,0 +1,4 @@
{
"schema_version": "coordination-frame/v1",
"frame_seconds": 3600
}

View file

@ -0,0 +1,31 @@
# Ecosystem coordination frame
Founder decision in the NK-WP-0035 sitting, 2026-10-15: use a configurable base
frame for long-term ecosystem coordination, initially **f = 1 hour**, replacing
the daily pace. Authoritative configuration: `config/coordination-frame.json`.
`frame_seconds` is the one operator setting. `coordination_engine.timing`
validates it and derives durations and timezone-bound deadlines in frames.
```sh
PYTHONPATH=src python3 -m coordination_engine.timing --config config/coordination-frame.json
```
Coordination wait stages are expressed as 2f and 3f. Owners adopt the frame
explicitly and retain the configuration digest in generated artifacts. The
first adoption is NK-WP-0035: source heartbeat f, receiver registration 2f,
then Audit Core's existing 1.5 grace gives an effective deadline of 3f.
At the current setting those durations are 1h, 2h and 3h. Changing f to 24h
reproduces the former 24h/48h/72h cadence without changing the multipliers.
NetKingdom's `tools/coordination-frame/render.py` checks or regenerates the
first source/runtime bindings. Generated consumer values are compiled copies
of this configuration, with provenance; they are not independent knobs.
Runtime owners review and deploy those artifacts through their normal change
paths. A configuration edit or a local deadline calculation does not enable
workers, apply a deployment, change a credential lifetime, or establish that
any other ecosystem timer has adopted this frame. New owners must name their
multipliers and verify adoption rather than silently scaling every timeout.
Worker transport polling, leases, retry backoff, credential TTLs and approval
validity are separate owner contracts. This parameter paces coordination and
cadence; it grants no authority and does not extend security validity.

View file

@ -0,0 +1,79 @@
"""Owner-defined coordination frame; no worker, credential or policy side effects."""
from __future__ import annotations
import argparse
import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
def _unique(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError("duplicate coordination frame key")
result[key] = value
return result
@dataclass(frozen=True)
class CoordinationFrame:
frame_seconds: int
def __post_init__(self):
if type(self.frame_seconds) is not int or self.frame_seconds <= 0:
raise ValueError("frame_seconds must be a positive integer")
self.duration(3)
def seconds(self, frames: int) -> int:
if type(frames) is not int or frames <= 0:
raise ValueError("frames must be a positive integer")
return self.frame_seconds * frames
def duration(self, frames: int) -> timedelta:
try:
return timedelta(seconds=self.seconds(frames))
except OverflowError as exc:
raise ValueError("coordination frame exceeds datetime range") from exc
def deadline(self, anchor: datetime, frames: int) -> datetime:
if anchor.tzinfo is None or anchor.utcoffset() is None:
raise ValueError("deadline anchor must include a timezone")
return anchor + self.duration(frames)
def as_dict(self):
return {
"frame_seconds": self.frame_seconds,
"2f_seconds": self.seconds(2),
"3f_seconds": self.seconds(3),
}
def load_frame(path: Path) -> CoordinationFrame:
document = json.loads(path.read_text(), object_pairs_hook=_unique)
if not isinstance(document, dict) or set(document) != {"schema_version", "frame_seconds"}:
raise ValueError("coordination frame requires only schema_version and frame_seconds")
if document["schema_version"] != "coordination-frame/v1":
raise ValueError("unsupported coordination frame schema")
return CoordinationFrame(document["frame_seconds"])
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--config", type=Path, required=True)
args = parser.parse_args(argv)
try:
frame = load_frame(args.config)
payload = frame.as_dict()
payload["config_sha256"] = hashlib.sha256(args.config.read_bytes()).hexdigest()
except (OSError, ValueError) as exc:
parser.error(str(exc))
print(json.dumps(payload, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())

59
tests/test_timing.py Normal file
View file

@ -0,0 +1,59 @@
import json
from datetime import datetime, timezone
import pytest
from coordination_engine.timing import CoordinationFrame, load_frame
def test_one_hour_frame_sets_two_and_three_hour_deadlines():
frame = CoordinationFrame(3600)
anchor = datetime(2026, 10, 15, 12, tzinfo=timezone.utc)
assert frame.deadline(anchor, 2).hour == 14
assert frame.deadline(anchor, 3).hour == 15
def test_frame_change_rescales_all_multiples():
assert CoordinationFrame(86400).as_dict() == {
"frame_seconds": 86400,
"2f_seconds": 172800,
"3f_seconds": 259200,
}
assert CoordinationFrame(600).as_dict() == {
"frame_seconds": 600,
"2f_seconds": 1200,
"3f_seconds": 1800,
}
@pytest.mark.parametrize("value", [0, -1, True, 1.5, "3600", None])
def test_invalid_base_frame_is_refused(value):
with pytest.raises(ValueError):
CoordinationFrame(value)
def test_naive_deadline_anchor_is_refused():
with pytest.raises(ValueError, match="timezone"):
CoordinationFrame(3600).deadline(datetime(2026, 10, 15), 2)
@pytest.mark.parametrize(
"raw",
[
'{"schema_version":"coordination-frame/v1","frame_seconds":3600,"frame_seconds":24}',
'{"schema_version":"other","frame_seconds":3600}',
'{"schema_version":"coordination-frame/v1","frame_seconds":3600,"ttl":5}',
"[]",
"{}",
],
)
def test_invalid_configuration_is_refused_without_fallback(tmp_path, raw):
path = tmp_path / "frame.json"
path.write_text(raw)
with pytest.raises(ValueError):
load_frame(path)
def test_explicit_config_is_read_without_wall_clock_or_environment_defaults(tmp_path):
path = tmp_path / "frame.json"
path.write_text(json.dumps({"schema_version": "coordination-frame/v1", "frame_seconds": 3600}))
assert load_frame(path).seconds(2) == 7200