Relay / fictional worked example
Relay contract-driven intake — runnable teaching slice
On this page
Download source, contracts & tests ↓
Python 3.10+ · standard library only · includes instructions and verification results.
This small Python package demonstrates one phase of the fictional Relay product: a request intake boundary with an explicit idempotency contract, an audit event and repeatable tests.
It deliberately has no external dependencies and no framework. It is an educational example, not a production implementation, and it does not implement assignment, notification, authentication, tenancy or release operations.
Run the checks
From this directory:
python3 -m unittest discover -s tests -v
python3 tools/validate_fixture.py fixtures/valid-request.json
python3 -m compileall -q relay tools testsThe fixture validator returns nonzero for an invalid fixture:
python3 tools/validate_fixture.py fixtures/invalid-request.json
# expected: exit status 1 with an error messageFiles
contracts/request.schema.json— canonical record shape for this example.relay/models.py— immutable request record, errors and canonical input validation.relay/store.py— in-memory mechanics with idempotent creation.relay/intake.py— integration that emits creation audit only on first submission.relay/audit.py— small audit event and sink.tests/test_intake.py— contract behaviour checks.tests/test_fixture_tool.py— tests the deterministic fixture validator.tools/validate_fixture.py— basic schema-subset validator, not a full JSON Schema engine.TEST-RESULTS.md— records checks actually run while assembling this pack.
What the tests establish
Within this in-memory demonstration: identical replay gives the original record; a conflicting replay is rejected; invalid input is rejected before writing; creation audit is not duplicated on exact replay; independent submissions can coexist.
What the tests do NOT establish
- Durable storage, cross-process uniqueness, concurrency across machines.
- Atomic persistence of request and audit event.
- Authentication, authorisation, tenant isolation, encryption or compliance.
- Correct product value, usability or market fit.
- Deployability or operational supportability.
In fact, a deliberate test highlights that if the audit sink fails after the in-memory store creates a request, the request can remain without an audit event. This is a known limitation and a reason to consider an outbox/transaction boundary in a real production design.
Why include code?
Because a technical principle should point to a falsifiable behavioural contract. An agent’s explanation that it implemented “duplicate protection” is weaker than repeatable checks covering exact replay, conflicting payloads and side effects.
Read the source and verification record
These are the same files included in the download. Expand any file to inspect it.
contracts/request.schema.json
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://example.invalid/relay/request.schema.json",
"title": "Relay request (teaching example)",
"type": "object",
"additionalProperties": false,
"required": ["id", "source", "external_id", "site_id", "summary", "status", "assignee_id", "version"],
"properties": {
"id": {"type": "string", "minLength": 1},
"source": {"type": "string", "minLength": 1},
"external_id": {"type": "string", "minLength": 1},
"site_id": {"type": "string", "minLength": 1},
"summary": {"type": "string", "minLength": 1},
"status": {"type": "string", "enum": ["new", "assigned", "resolved"]},
"assignee_id": {"type": ["string", "null"]},
"version": {"type": "integer", "minimum": 1}
}
}
fixtures/invalid-request.json
{
"id": "req-000001",
"source": "site-form",
"external_id": "submission-42",
"site_id": "site-a",
"summary": "Leaking tap in kitchen",
"status": "new",
"assignee_id": null,
"version": 0
}
fixtures/valid-request.json
{
"id": "req-000001",
"source": "site-form",
"external_id": "submission-42",
"site_id": "site-a",
"summary": "Leaking tap in kitchen",
"status": "new",
"assignee_id": null,
"version": 1
}
relay/__init__.py
"""Relay intake demonstration: not production software."""
from .audit import AuditEvent, InMemoryAuditSink
from .intake import RequestIntake, SubmissionResult
from .models import DuplicatePayloadConflict, Request, ValidationError
from .store import InMemoryRequestStore
__all__ = [
"AuditEvent", "InMemoryAuditSink", "RequestIntake", "SubmissionResult",
"DuplicatePayloadConflict", "Request", "ValidationError", "InMemoryRequestStore",
]
relay/audit.py
"""Audit port used by the intake integration."""
from dataclasses import dataclass
from typing import Protocol
@dataclass(frozen=True)
class AuditEvent:
kind: str
request_id: str
source: str
external_id: str
class AuditSink(Protocol):
def append(self, event: AuditEvent) -> None: ...
class InMemoryAuditSink:
def __init__(self) -> None:
self.events: list[AuditEvent] = []
def append(self, event: AuditEvent) -> None:
self.events.append(event)
relay/intake.py
"""Integration composes mechanics without re-implementing store policy."""
from dataclasses import dataclass
from .audit import AuditEvent, AuditSink
from .models import Request
from .store import RequestStore
@dataclass(frozen=True)
class SubmissionResult:
request: Request
created: bool
class RequestIntake:
def __init__(self, store: RequestStore, audit: AuditSink) -> None:
self._store = store
self._audit = audit
def submit(
self, *, source: object, external_id: object, site_id: object, summary: object
) -> SubmissionResult:
request, created = self._store.create_if_absent(
source=source,
external_id=external_id,
site_id=site_id,
summary=summary,
)
if created:
# Teaching limitation: store + audit are NOT atomic here.
self._audit.append(
AuditEvent(
kind="RequestCreated",
request_id=request.id,
source=request.source,
external_id=request.external_id,
)
)
return SubmissionResult(request=request, created=created)
relay/models.py
"""Immutable contract types and canonical input validation."""
from dataclasses import dataclass
class ValidationError(ValueError):
"""Request input violates the agreed intake contract."""
class DuplicatePayloadConflict(ValueError):
"""One external key has been presented with contradictory content."""
@dataclass(frozen=True)
class Request:
id: str
source: str
external_id: str
site_id: str
summary: str
status: str = "new"
assignee_id: str | None = None
version: int = 1
def _required_text(value: object, field: str) -> str:
if not isinstance(value, str) or not value.strip():
raise ValidationError(f"{field} must be a non-empty string")
return value.strip()
def canonical_input(
source: object,
external_id: object,
site_id: object,
summary: object,
) -> tuple[str, str, str, str]:
"""Trim surrounding whitespace; do not casefold or apply fuzzy matching."""
return (
_required_text(source, "source"),
_required_text(external_id, "external_id"),
_required_text(site_id, "site_id"),
_required_text(summary, "summary"),
)
relay/store.py
"""Single-process in-memory mechanics for demonstrating the store contract."""
from threading import RLock
from typing import Protocol
from .models import DuplicatePayloadConflict, Request, canonical_input
class RequestStore(Protocol):
"""Consumer-facing port; the integration must not depend on a storage adapter."""
def create_if_absent(
self, *, source: object, external_id: object,
site_id: object, summary: object
) -> tuple[Request, bool]: ...
class InMemoryRequestStore:
def __init__(self) -> None:
self._by_external_key: dict[tuple[str, str], Request] = {}
self._sequence = 0
self._lock = RLock()
def create_if_absent(
self,
*,
source: object,
external_id: object,
site_id: object,
summary: object,
) -> tuple[Request, bool]:
source, external_id, site_id, summary = canonical_input(
source, external_id, site_id, summary
)
key = (source, external_id)
with self._lock:
existing = self._by_external_key.get(key)
if existing is not None:
if (existing.site_id, existing.summary) != (site_id, summary):
raise DuplicatePayloadConflict(
"external key already exists with a different payload"
)
return existing, False
self._sequence += 1
request = Request(
id=f"req-{self._sequence:06d}",
source=source,
external_id=external_id,
site_id=site_id,
summary=summary,
)
self._by_external_key[key] = request
return request, True
def count(self) -> int:
with self._lock:
return len(self._by_external_key)
def get_by_external_key(self, source: str, external_id: str) -> Request | None:
with self._lock:
return self._by_external_key.get((source, external_id))
tests/test_fixture_tool.py
"""Check that the repeatable fixture-validation tool behaves predictably."""
import subprocess
import sys
import unittest
from pathlib import Path
BASE = Path(__file__).resolve().parents[1]
TOOL = BASE / "tools" / "validate_fixture.py"
class FixtureValidatorTests(unittest.TestCase):
def run_tool(self, fixture: str) -> subprocess.CompletedProcess[str]:
return subprocess.run(
[sys.executable, str(TOOL), str(BASE / "fixtures" / fixture)],
text=True,
capture_output=True,
check=False,
)
def test_valid_fixture_passes(self) -> None:
result = self.run_tool("valid-request.json")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("VALID:", result.stdout)
def test_invalid_fixture_fails_with_reason(self) -> None:
result = self.run_tool("invalid-request.json")
self.assertEqual(result.returncode, 1)
self.assertIn("version: must be >= 1", result.stderr)
if __name__ == "__main__":
unittest.main()
tests/test_intake.py
"""Executable behavioural-contract tests for the Relay intake slice."""
import unittest
from relay import (
DuplicatePayloadConflict,
InMemoryAuditSink,
InMemoryRequestStore,
RequestIntake,
ValidationError,
)
class RequestIntakeTests(unittest.TestCase):
def setUp(self) -> None:
self.store = InMemoryRequestStore()
self.audit = InMemoryAuditSink()
self.intake = RequestIntake(self.store, self.audit)
self.payload = dict(
source="site-form",
external_id="submission-42",
site_id="site-a",
summary="Leaking tap in kitchen",
)
def test_new_submission_creates_record_with_version_one(self) -> None:
result = self.intake.submit(**self.payload)
self.assertTrue(result.created)
self.assertEqual(result.request.version, 1)
self.assertEqual(result.request.status, "new")
self.assertEqual(result.request.assignee_id, None)
self.assertEqual(self.store.count(), 1)
def test_exact_replay_returns_original(self) -> None:
first = self.intake.submit(**self.payload)
second = self.intake.submit(**self.payload)
self.assertFalse(second.created)
self.assertEqual(second.request, first.request)
self.assertEqual(self.store.count(), 1)
def test_whitespace_trim_is_canonical_input(self) -> None:
first = self.intake.submit(**self.payload)
padded = dict(self.payload)
padded["summary"] = " Leaking tap in kitchen "
second = self.intake.submit(**padded)
self.assertFalse(second.created)
self.assertEqual(second.request, first.request)
def test_conflicting_duplicate_rejected_without_mutation(self) -> None:
first = self.intake.submit(**self.payload)
changed = dict(self.payload)
changed["summary"] = "Different issue"
with self.assertRaises(DuplicatePayloadConflict):
self.intake.submit(**changed)
self.assertEqual(self.store.count(), 1)
self.assertEqual(
self.store.get_by_external_key("site-form", "submission-42"),
first.request,
)
self.assertEqual(len(self.audit.events), 1)
def test_site_conflict_is_not_silently_accepted(self) -> None:
self.intake.submit(**self.payload)
changed = dict(self.payload)
changed["site_id"] = "site-b"
with self.assertRaises(DuplicatePayloadConflict):
self.intake.submit(**changed)
def test_invalid_summary_has_no_side_effects(self) -> None:
changed = dict(self.payload)
changed["summary"] = " "
with self.assertRaises(ValidationError):
self.intake.submit(**changed)
self.assertEqual(self.store.count(), 0)
self.assertEqual(len(self.audit.events), 0)
def test_invalid_non_string_key_is_rejected(self) -> None:
changed = dict(self.payload)
changed["external_id"] = 42
with self.assertRaises(ValidationError):
self.intake.submit(**changed)
self.assertEqual(self.store.count(), 0)
def test_same_summary_distinct_external_keys_are_distinct_requests(self) -> None:
first = self.intake.submit(**self.payload)
another = dict(self.payload)
another["external_id"] = "submission-43"
second = self.intake.submit(**another)
self.assertNotEqual(first.request.id, second.request.id)
self.assertEqual(self.store.count(), 2)
def test_different_sources_namespace_external_ids(self) -> None:
self.intake.submit(**self.payload)
another = dict(self.payload)
another["source"] = "email-gateway"
self.intake.submit(**another)
self.assertEqual(self.store.count(), 2)
def test_creation_audit_is_emitted_only_once(self) -> None:
self.intake.submit(**self.payload)
self.intake.submit(**self.payload)
self.assertEqual(len(self.audit.events), 1)
self.assertEqual(self.audit.events[0].kind, "RequestCreated")
def test_audit_failure_exposes_production_transaction_gap(self) -> None:
class FailingAudit:
def append(self, event):
raise RuntimeError("simulated audit outage")
intake = RequestIntake(self.store, FailingAudit())
with self.assertRaises(RuntimeError):
intake.submit(**self.payload)
# This intentionally illustrates why a durable outbox/transaction is needed.
self.assertEqual(self.store.count(), 1)
if __name__ == "__main__":
unittest.main()
tools/validate_fixture.py
#!/usr/bin/env python3
"""Small, deterministic validator for the specific schema features used here.
Not a general-purpose JSON Schema implementation. Uses no external packages.
"""
import json
import sys
from pathlib import Path
SCHEMA = Path(__file__).resolve().parents[1] / "contracts" / "request.schema.json"
def _is_type(value: object, kind: str) -> bool:
if kind == "null":
return value is None
if kind == "string":
return isinstance(value, str)
if kind == "integer":
return isinstance(value, int) and not isinstance(value, bool)
if kind == "object":
return isinstance(value, dict)
raise ValueError(f"Unsupported schema type: {kind}")
def validate(data: object, schema: dict) -> list[str]:
issues: list[str] = []
if not isinstance(data, dict):
return ["request must be a JSON object"]
props = schema["properties"]
for field in schema["required"]:
if field not in data:
issues.append(f"missing required field: {field}")
if schema.get("additionalProperties") is False:
for field in data:
if field not in props:
issues.append(f"unknown field: {field}")
for field, value in data.items():
if field not in props:
continue
rule = props[field]
kinds = rule.get("type", [])
if isinstance(kinds, str):
kinds = [kinds]
if not any(_is_type(value, kind) for kind in kinds):
issues.append(f"{field}: expected {' or '.join(kinds)}")
continue
if "enum" in rule and value not in rule["enum"]:
issues.append(f"{field}: invalid enum value {value!r}")
if isinstance(value, str) and "minLength" in rule:
if len(value) < rule["minLength"] or not value.strip():
issues.append(f"{field}: must be non-empty")
if isinstance(value, int) and not isinstance(value, bool) and "minimum" in rule:
if value < rule["minimum"]:
issues.append(f"{field}: must be >= {rule['minimum']}")
return issues
def main(args: list[str]) -> int:
if len(args) != 1:
print("Usage: validate_fixture.py <fixture.json>", file=sys.stderr)
return 2
try:
schema = json.loads(SCHEMA.read_text(encoding="utf-8"))
data = json.loads(Path(args[0]).read_text(encoding="utf-8"))
issues = validate(data, schema)
except (OSError, json.JSONDecodeError, ValueError) as err:
print(f"Fixture validation error: {err}", file=sys.stderr)
return 2
if issues:
for issue in issues:
print(f"INVALID: {issue}", file=sys.stderr)
return 1
print(f"VALID: {args[0]}")
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))
VERIFICATION.md
# Verification in the website repository
Re-run using Python 3.13.15.
These checks establish the limited in-memory teaching behaviours, not production readiness or product-market validation. The audit-failure test demonstrates a known transaction gap.
```text
$ python -m unittest discover -s tests -v
test_invalid_fixture_fails_with_reason (test_fixture_tool.FixtureValidatorTests.test_invalid_fixture_fails_with_reason) ... ok
test_valid_fixture_passes (test_fixture_tool.FixtureValidatorTests.test_valid_fixture_passes) ... ok
test_audit_failure_exposes_production_transaction_gap (test_intake.RequestIntakeTests.test_audit_failure_exposes_production_transaction_gap) ... ok
test_conflicting_duplicate_rejected_without_mutation (test_intake.RequestIntakeTests.test_conflicting_duplicate_rejected_without_mutation) ... ok
test_creation_audit_is_emitted_only_once (test_intake.RequestIntakeTests.test_creation_audit_is_emitted_only_once) ... ok
test_different_sources_namespace_external_ids (test_intake.RequestIntakeTests.test_different_sources_namespace_external_ids) ... ok
test_exact_replay_returns_original (test_intake.RequestIntakeTests.test_exact_replay_returns_original) ... ok
test_invalid_non_string_key_is_rejected (test_intake.RequestIntakeTests.test_invalid_non_string_key_is_rejected) ... ok
test_invalid_summary_has_no_side_effects (test_intake.RequestIntakeTests.test_invalid_summary_has_no_side_effects) ... ok
test_new_submission_creates_record_with_version_one (test_intake.RequestIntakeTests.test_new_submission_creates_record_with_version_one) ... ok
test_same_summary_distinct_external_keys_are_distinct_requests (test_intake.RequestIntakeTests.test_same_summary_distinct_external_keys_are_distinct_requests) ... ok
test_site_conflict_is_not_silently_accepted (test_intake.RequestIntakeTests.test_site_conflict_is_not_silently_accepted) ... ok
test_whitespace_trim_is_canonical_input (test_intake.RequestIntakeTests.test_whitespace_trim_is_canonical_input) ... ok
----------------------------------------------------------------------
Ran 13 tests in 0.128s
OK
Exit status: 0 (expected 0)
$ python tools/validate_fixture.py fixtures/valid-request.json
VALID: fixtures/valid-request.json
Exit status: 0 (expected 0)
$ python tools/validate_fixture.py fixtures/invalid-request.json
INVALID: version: must be >= 1
Exit status: 1 (expected 1)
$ python -m compileall -q relay tools tests
Exit status: 0 (expected 0)
```