coulomb/tenant-engine
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

TEN-WP-0002 T01-T03: service skeleton, domain model, storage layer

Browse source 
Python 3.12 + FastAPI, pyproject.toml + Makefile mirroring
qonto-assistant's exactly. src/tenant_engine/{domain,store,app,main}.py:

- domain.py: Tenant, CapabilityRole (PLTF/IAM/VEN/CUS), RoleGrant,
  PlanAssignment, create_role_grant() enforcing ADR-0014's invariants.
  Refinement made while implementing: platform_default grants are valid
  for trial-grouped tenants OR the reserved tenant:platform/tenant:coulomb
  tenants (their baseline roles were never purchased either) -- the task
  spec only named the trial case.
- store.py: TenantStore Protocol + InMemoryTenantStore, every mutation
  emits a DomainEvent per the boundary contract's Audit Correlation
  Contract.
- app.py/main.py: FastAPI factory + /health endpoint, verified live on
  127.0.0.1:8090.

29 tests passing, including non-exclusive role coexistence (CUS+VEN
simultaneously) and append-only revoke semantics.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-07-23 22:01:23 +02:00
parent 7eb21c05b8
commit 0770ce82d9
11 changed files with 774 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

33
Makefile Normal file
View file

@ -0,0 +1,33 @@
VENV ?= .venv
PYTHON ?= $(VENV)/bin/python
PIP ?= $(PYTHON) -m pip
PYTEST ?= $(PYTHON) -m pytest
RUFF ?= $(VENV)/bin/ruff
.PHONY: help install-dev test lint run
help:
@echo "make install-dev Create .venv and install runtime + dev dependencies"
@echo "make test Run the test suite"
@echo "make lint Run syntax and static checks"
@echo "make run Start the local API on 127.0.0.1:8090"
$(VENV)/bin/python:
python3 -m venv $(VENV)
$(VENV)/.dev-installed: pyproject.toml $(VENV)/bin/python
$(PIP) install --upgrade pip
$(PIP) install -e ".[dev]"
@touch $(VENV)/.dev-installed
install-dev: $(VENV)/.dev-installed
test: $(VENV)/.dev-installed
$(PYTEST)
lint: $(VENV)/.dev-installed
$(PYTHON) -m compileall src tests
$(RUFF) check src tests
run: $(VENV)/.dev-installed
$(PYTHON) -m tenant_engine.main

44
pyproject.toml Normal file
View file

@ -0,0 +1,44 @@
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "tenant-engine"
version = "0.1.0"
description = "Canonical owner of tenant-as-an-entity facts for NetKingdom: existence, onboarding grouping, capability roles, and plan/subscription assignment."
readme = "README.md"
requires-python = ">=3.12"
license = { file = "LICENSE" }
authors = [{ name = "Coulomb" }]
dependencies = [
"fastapi>=0.115,<1.0",
"uvicorn[standard]>=0.30,<1.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.2,<9.0",
"httpx>=0.27,<1.0",
"ruff>=0.6,<1.0",
]
[project.scripts]
tenant-engine = "tenant_engine.main:main"
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = [
"--strict-markers",
"--disable-warnings",
"--tb=short",
]
[tool.ruff]
line-length = 100
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "I", "B"]

1
src/tenant_engine/__init__.py Normal file
View file

@ -0,0 +1 @@
__version__ = "0.1.0"

19
src/tenant_engine/app.py Normal file
View file

@ -0,0 +1,19 @@
from __future__ import annotations
from fastapi import FastAPI
from tenant_engine import __version__
from tenant_engine.store import InMemoryTenantStore, TenantStore
def create_app(*, store: TenantStore | None = None) -> FastAPI:
store = store or InMemoryTenantStore()
app = FastAPI(title="tenant-engine", version=__version__)
app.state.store = store
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok", "service": "tenant-engine", "version": __version__}
return app

170
src/tenant_engine/domain.py Normal file
View file

@ -0,0 +1,170 @@
from __future__ import annotations
from dataclasses import dataclass, replace
from datetime import datetime
from enum import Enum
from typing import Literal
# ADR-0013: onboarding-risk / entity-shape grouping. Orthogonal to
# capability role (below) -- neither axis constrains the other except
# through guardrail policy, which is a reserved, unimplemented concern.
GROUPINGS = frozenset(
{
"trial",
"friendly",
"single",
"small",
"medium",
"large",
"enterprise",
"consumer",
"family",
"community",
"association",
"agentic",
}
)
# tenant:platform and tenant:coulomb predate the grouping taxonomy and stay
# reserved, ungrouped identifiers (ADR-0013's Tenant Claim rationale).
RESERVED_IDENTIFIERS = frozenset({"tenant:platform", "tenant:coulomb"})
class InvalidTenantIdentifierError(ValueError):
"""A tenant identifier does not match `tenant:<grouping>:<name>` or a reserved form."""
class InvalidGrantError(ValueError):
"""A role grant violates a domain invariant (ADR-0014)."""
class CapabilityRole(str, Enum):
"""ADR-0014: non-exclusive capability roles a tenant may hold."""
PLTF = "PLTF"
IAM = "IAM"
VEN = "VEN"
CUS = "CUS"
GrantReason = Literal["plan_assignment", "manual_grant", "platform_default"]
def parse_tenant_identifier(identifier: str) -> tuple[str | None, str]:
"""Return (grouping, name). grouping is None only for reserved identifiers."""
if identifier in RESERVED_IDENTIFIERS:
return None, identifier.split(":", 1)[1]
parts = identifier.split(":")
if len(parts) != 3 or parts[0] != "tenant":
raise InvalidTenantIdentifierError(
f"Malformed tenant identifier: {identifier!r} (expected tenant:<grouping>:<name>)"
)
_, grouping, name = parts
if grouping not in GROUPINGS:
raise InvalidTenantIdentifierError(f"Unknown tenant grouping: {grouping!r}")
if not name:
raise InvalidTenantIdentifierError("Tenant name segment is empty")
return grouping, name
@dataclass(frozen=True, slots=True)
class Tenant:
tenant_id: str
identifier: str
grouping: str | None
@classmethod
def create(cls, *, tenant_id: str, identifier: str) -> "Tenant":
grouping, _name = parse_tenant_identifier(identifier)
return cls(tenant_id=tenant_id, identifier=identifier, grouping=grouping)
@property
def is_reserved(self) -> bool:
return self.grouping is None
@dataclass(frozen=True, slots=True)
class RoleGrant:
"""Audited role grant record -- the Tenant Role & Plan Grant Contract shape.
Append-only: revocation produces a new record via `revoke()`, it never
deletes the original.
"""
grant_id: str
tenant_id: str
role: CapabilityRole
grant_reason: GrantReason
plan_id: str | None
granted_by: str
granted_at: datetime
correlation_id: str
revoked_at: datetime | None = None
@property
def active(self) -> bool:
return self.revoked_at is None
def revoke(self, *, at: datetime) -> "RoleGrant":
if self.revoked_at is not None:
raise InvalidGrantError(f"Grant {self.grant_id!r} is already revoked")
return replace(self, revoked_at=at)
@dataclass(frozen=True, slots=True)
class PlanAssignment:
"""A tenant's current plan, referenced by id only -- never resolved or
duplicated locally. Plan term definitions belong to adaptive-pricing.
"""
tenant_id: str
plan_id: str
assigned_at: datetime
def create_role_grant(
*,
tenant: Tenant,
grant_id: str,
role: CapabilityRole,
grant_reason: GrantReason,
plan_id: str | None,
granted_by: str,
correlation_id: str,
granted_at: datetime,
) -> RoleGrant:
"""Construct a RoleGrant, enforcing ADR-0014's domain invariants.
- `plan_assignment` grants always require a `plan_id`.
- `platform_default` grants never carry a `plan_id`, and are only valid
for `trial`-grouped tenants (ADR-0014: trial may hold any role,
unrestricted, for showcase purposes) or the reserved, ungrouped
`tenant:platform`/`tenant:coulomb` tenants (their baseline roles were
never purchased either).
- `manual_grant` carries no grouping restriction and an optional
`plan_id`.
"""
if grant_reason == "plan_assignment" and plan_id is None:
raise InvalidGrantError("plan_assignment grants require a plan_id")
if grant_reason == "platform_default":
if plan_id is not None:
raise InvalidGrantError("platform_default grants must not carry a plan_id")
if tenant.grouping not in (None, "trial"):
raise InvalidGrantError(
"platform_default is only valid for trial-grouped or reserved tenants, "
f"got grouping={tenant.grouping!r}"
)
return RoleGrant(
grant_id=grant_id,
tenant_id=tenant.tenant_id,
role=role,
grant_reason=grant_reason,
plan_id=plan_id,
granted_by=granted_by,
granted_at=granted_at,
correlation_id=correlation_id,
)

13
src/tenant_engine/main.py Normal file
View file

@ -0,0 +1,13 @@
from __future__ import annotations
import uvicorn
from tenant_engine.app import create_app
def main() -> None:
uvicorn.run(create_app(), host="127.0.0.1", port=8090)
if __name__ == "__main__":
main()

120
src/tenant_engine/store.py Normal file
View file

@ -0,0 +1,120 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any, Protocol
from tenant_engine.domain import CapabilityRole, PlanAssignment, RoleGrant, Tenant
class TenantNotFoundError(KeyError):
pass
class TenantAlreadyExistsError(ValueError):
pass
class GrantNotFoundError(KeyError):
pass
@dataclass(frozen=True, slots=True)
class DomainEvent:
"""Boundary contract's Audit Correlation Contract, in event form."""
event_type: str
tenant_id: str
at: datetime
payload: dict[str, Any]
class TenantStore(Protocol):
"""Swappable persistence seam -- domain/ and api/ depend on this, not a backend."""
def create_tenant(self, tenant: Tenant) -> None: ...
def get_tenant(self, tenant_id: str) -> Tenant: ...
def grant_role(self, grant: RoleGrant) -> None: ...
def revoke_role(self, *, tenant_id: str, grant_id: str, at: datetime) -> RoleGrant: ...
def active_roles(self, tenant_id: str) -> frozenset[CapabilityRole]: ...
def assign_plan(self, assignment: PlanAssignment) -> None: ...
def events(self) -> list[DomainEvent]: ...
class InMemoryTenantStore:
def __init__(self) -> None:
self._tenants: dict[str, Tenant] = {}
self._grants: dict[str, dict[str, RoleGrant]] = {}
self._plans: dict[str, PlanAssignment] = {}
self._events: list[DomainEvent] = []
def create_tenant(self, tenant: Tenant) -> None:
if tenant.tenant_id in self._tenants:
raise TenantAlreadyExistsError(tenant.tenant_id)
self._tenants[tenant.tenant_id] = tenant
self._grants[tenant.tenant_id] = {}
self._emit(
"tenant_created",
tenant.tenant_id,
{"identifier": tenant.identifier, "grouping": tenant.grouping},
)
def get_tenant(self, tenant_id: str) -> Tenant:
try:
return self._tenants[tenant_id]
except KeyError:
raise TenantNotFoundError(tenant_id) from None
def grant_role(self, grant: RoleGrant) -> None:
self.get_tenant(grant.tenant_id)
self._grants[grant.tenant_id][grant.grant_id] = grant
self._emit(
"role_granted",
grant.tenant_id,
{
"grant_id": grant.grant_id,
"role": grant.role.value,
"grant_reason": grant.grant_reason,
"correlation_id": grant.correlation_id,
},
)
def revoke_role(self, *, tenant_id: str, grant_id: str, at: datetime) -> RoleGrant:
self.get_tenant(tenant_id)
try:
grant = self._grants[tenant_id][grant_id]
except KeyError:
raise GrantNotFoundError(grant_id) from None
revoked = grant.revoke(at=at)
self._grants[tenant_id][grant_id] = revoked
self._emit(
"role_revoked",
tenant_id,
{"grant_id": grant_id, "role": revoked.role.value},
)
return revoked
def active_roles(self, tenant_id: str) -> frozenset[CapabilityRole]:
self.get_tenant(tenant_id)
return frozenset(
grant.role for grant in self._grants.get(tenant_id, {}).values() if grant.active
)
def assign_plan(self, assignment: PlanAssignment) -> None:
self.get_tenant(assignment.tenant_id)
self._plans[assignment.tenant_id] = assignment
self._emit("plan_assigned", assignment.tenant_id, {"plan_id": assignment.plan_id})
def events(self) -> list[DomainEvent]:
return list(self._events)
def _emit(self, event_type: str, tenant_id: str, payload: dict[str, Any]) -> None:
self._events.append(
DomainEvent(event_type=event_type, tenant_id=tenant_id, at=datetime.now(UTC), payload=payload)
)

13
tests/test_app.py Normal file
View file

@ -0,0 +1,13 @@
from fastapi.testclient import TestClient
from tenant_engine.app import create_app
def test_health_endpoint() -> None:
client = TestClient(create_app())
response = client.get("/health")
assert response.status_code == 200
body = response.json()
assert body["status"] == "ok"
assert body["service"] == "tenant-engine"

206
tests/test_domain.py Normal file
View file

@ -0,0 +1,206 @@
from datetime import UTC, datetime
import pytest
from tenant_engine.domain import (
CapabilityRole,
InvalidGrantError,
InvalidTenantIdentifierError,
Tenant,
create_role_grant,
parse_tenant_identifier,
)
def test_parse_tenant_identifier_reserved_platform() -> None:
grouping, name = parse_tenant_identifier("tenant:platform")
assert grouping is None
assert name == "platform"
def test_parse_tenant_identifier_reserved_coulomb() -> None:
grouping, name = parse_tenant_identifier("tenant:coulomb")
assert grouping is None
assert name == "coulomb"
def test_parse_tenant_identifier_grouped() -> None:
grouping, name = parse_tenant_identifier("tenant:friendly:binky")
assert grouping == "friendly"
assert name == "binky"
@pytest.mark.parametrize(
"identifier",
[
"tenant:unknown-grouping:binky",
"tenant:friendly",
"tenant:friendly:",
"not-a-tenant:friendly:binky",
"friendly:binky",
],
)
def test_parse_tenant_identifier_rejects_invalid_shapes(identifier: str) -> None:
with pytest.raises(InvalidTenantIdentifierError):
parse_tenant_identifier(identifier)
def test_tenant_create_from_identifier() -> None:
tenant = Tenant.create(tenant_id="t-1", identifier="tenant:friendly:binky")
assert tenant.grouping == "friendly"
assert tenant.is_reserved is False
def test_tenant_create_reserved_has_no_grouping() -> None:
tenant = Tenant.create(tenant_id="t-platform", identifier="tenant:platform")
assert tenant.grouping is None
assert tenant.is_reserved is True
def _tenant(*, grouping: str | None, identifier: str | None = None) -> Tenant:
if identifier is None:
identifier = f"tenant:{grouping}:acme" if grouping else "tenant:platform"
return Tenant.create(tenant_id="t-1", identifier=identifier)
def test_plan_assignment_grant_requires_plan_id() -> None:
tenant = _tenant(grouping="friendly")
with pytest.raises(InvalidGrantError):
create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.IAM,
grant_reason="plan_assignment",
plan_id=None,
granted_by="ops",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
def test_plan_assignment_grant_with_plan_id_succeeds() -> None:
tenant = _tenant(grouping="friendly")
grant = create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.IAM,
grant_reason="plan_assignment",
plan_id="plan-iam-dedicated",
granted_by="ops",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
assert grant.active is True
assert grant.plan_id == "plan-iam-dedicated"
def test_platform_default_allowed_for_trial_tenant_without_plan() -> None:
tenant = _tenant(grouping="trial")
grant = create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.VEN,
grant_reason="platform_default",
plan_id=None,
granted_by="platform",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
assert grant.grant_reason == "platform_default"
assert grant.plan_id is None
def test_platform_default_allowed_for_reserved_tenant() -> None:
tenant = _tenant(grouping=None, identifier="tenant:platform")
grant = create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.PLTF,
grant_reason="platform_default",
plan_id=None,
granted_by="platform",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
assert grant.role is CapabilityRole.PLTF
def test_platform_default_rejected_for_non_trial_grouped_tenant() -> None:
tenant = _tenant(grouping="enterprise")
with pytest.raises(InvalidGrantError):
create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.VEN,
grant_reason="platform_default",
plan_id=None,
granted_by="platform",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
def test_platform_default_rejected_when_plan_id_present() -> None:
tenant = _tenant(grouping="trial")
with pytest.raises(InvalidGrantError):
create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.VEN,
grant_reason="platform_default",
plan_id="plan-x",
granted_by="platform",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
def test_manual_grant_allows_missing_plan_id_for_any_grouping() -> None:
tenant = _tenant(grouping="enterprise")
grant = create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.CUS,
grant_reason="manual_grant",
plan_id=None,
granted_by="ops",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
assert grant.grant_reason == "manual_grant"
def test_grant_revoke_is_append_only() -> None:
tenant = _tenant(grouping="friendly")
grant = create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.CUS,
grant_reason="manual_grant",
plan_id=None,
granted_by="ops",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
revoked = grant.revoke(at=datetime.now(UTC))
assert grant.revoked_at is None, "original record must be untouched"
assert revoked.revoked_at is not None
assert revoked.grant_id == grant.grant_id
def test_revoking_twice_raises() -> None:
tenant = _tenant(grouping="friendly")
grant = create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.CUS,
grant_reason="manual_grant",
plan_id=None,
granted_by="ops",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
).revoke(at=datetime.now(UTC))
with pytest.raises(InvalidGrantError):
grant.revoke(at=datetime.now(UTC))

124
tests/test_store.py Normal file
View file

@ -0,0 +1,124 @@
from datetime import UTC, datetime
import pytest
from tenant_engine.domain import CapabilityRole, PlanAssignment, Tenant, create_role_grant
from tenant_engine.store import (
GrantNotFoundError,
InMemoryTenantStore,
TenantAlreadyExistsError,
TenantNotFoundError,
)
def _store_with_tenant(*, grouping: str = "friendly", name: str = "binky") -> tuple[InMemoryTenantStore, Tenant]:
store = InMemoryTenantStore()
tenant = Tenant.create(tenant_id=f"t-{name}", identifier=f"tenant:{grouping}:{name}")
store.create_tenant(tenant)
return store, tenant
def test_create_and_get_tenant() -> None:
store, tenant = _store_with_tenant()
assert store.get_tenant(tenant.tenant_id) == tenant
def test_create_duplicate_tenant_raises() -> None:
store, tenant = _store_with_tenant()
with pytest.raises(TenantAlreadyExistsError):
store.create_tenant(tenant)
def test_get_unknown_tenant_raises() -> None:
store = InMemoryTenantStore()
with pytest.raises(TenantNotFoundError):
store.get_tenant("does-not-exist")
def test_grant_and_active_roles() -> None:
store, tenant = _store_with_tenant()
grant = create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.CUS,
grant_reason="manual_grant",
plan_id=None,
granted_by="ops",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
store.grant_role(grant)
assert store.active_roles(tenant.tenant_id) == frozenset({CapabilityRole.CUS})
def test_revoke_removes_role_from_active_set_but_keeps_record() -> None:
store, tenant = _store_with_tenant()
grant = create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.VEN,
grant_reason="manual_grant",
plan_id=None,
granted_by="ops",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
store.grant_role(grant)
revoked = store.revoke_role(tenant_id=tenant.tenant_id, grant_id="g-1", at=datetime.now(UTC))
assert revoked.revoked_at is not None
assert store.active_roles(tenant.tenant_id) == frozenset()
def test_revoke_unknown_grant_raises() -> None:
store, tenant = _store_with_tenant()
with pytest.raises(GrantNotFoundError):
store.revoke_role(tenant_id=tenant.tenant_id, grant_id="missing", at=datetime.now(UTC))
def test_non_exclusive_roles_coexist() -> None:
store, tenant = _store_with_tenant()
for role in (CapabilityRole.CUS, CapabilityRole.VEN):
store.grant_role(
create_role_grant(
tenant=tenant,
grant_id=f"g-{role.value}",
role=role,
grant_reason="manual_grant",
plan_id=None,
granted_by="ops",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
)
assert store.active_roles(tenant.tenant_id) == frozenset({CapabilityRole.CUS, CapabilityRole.VEN})
def test_assign_plan() -> None:
store, tenant = _store_with_tenant()
store.assign_plan(PlanAssignment(tenant_id=tenant.tenant_id, plan_id="plan-x", assigned_at=datetime.now(UTC)))
events = store.events()
assert any(event.event_type == "plan_assigned" and event.payload["plan_id"] == "plan-x" for event in events)
def test_every_mutation_emits_an_event() -> None:
store, tenant = _store_with_tenant()
grant = create_role_grant(
tenant=tenant,
grant_id="g-1",
role=CapabilityRole.CUS,
grant_reason="manual_grant",
plan_id=None,
granted_by="ops",
correlation_id="corr-1",
granted_at=datetime.now(UTC),
)
store.grant_role(grant)
store.revoke_role(tenant_id=tenant.tenant_id, grant_id="g-1", at=datetime.now(UTC))
store.assign_plan(PlanAssignment(tenant_id=tenant.tenant_id, plan_id="plan-x", assigned_at=datetime.now(UTC)))
event_types = [event.event_type for event in store.events()]
assert event_types == ["tenant_created", "role_granted", "role_revoked", "plan_assigned"]

34
workplans/TEN-WP-0002-domain-model-and-scaffold.md
View file

@ -34,7 +34,7 @@ enforcement.
```task
id: TEN-WP-0002-T01
status: todo
status: done
priority: high
state_hub_task_id: "d22b0ce8-3f25-4c8d-b54f-ba44c65159c1"
```
@ -47,11 +47,19 @@ routers), from day one — not refactored in later.
Done when: `make test` runs an empty/smoke suite; `make run` starts a bare
FastAPI app with a `/health` endpoint.
**Done 2026-07-23:** `pyproject.toml` + `Makefile` land, mirroring
`qonto-assistant`'s exactly (`install-dev`/`test`/`lint`/`run`).
`src/tenant_engine/{domain,store,app,main}.py` scaffolded per the planned
layering (`domain.py` pure, `store.py` the persistence seam, `app.py` the
FastAPI factory). Verified live: `python -m tenant_engine.main` starts on
`127.0.0.1:8090`, `GET /health` returns
`{"status": "ok", "service": "tenant-engine", "version": "0.1.0"}`.
## Task: Domain model — tenant, grouping, capability role, plan grant
```task
id: TEN-WP-0002-T02
status: todo
status: done
priority: high
state_hub_task_id: "cfceea30-f383-4781-a7e4-226653aafca9"
```
@ -84,11 +92,23 @@ Validation rules encoded as domain invariants, not just API-layer checks:
Done when: unit tests cover valid/invalid grouping values, the trial/
no-plan-required exception, and grant/revoke as append-only operations.
**Done 2026-07-23:** `domain.py` implements `Tenant`, `CapabilityRole`,
`RoleGrant`, `PlanAssignment`, and `create_role_grant()` exactly as
specified. One refinement made while implementing, not pre-specified in the
task text: `platform_default` is valid for `trial`-grouped tenants **or**
the reserved, ungrouped `tenant:platform`/`tenant:coulomb` tenants (their
baseline roles were never purchased either) — the task only mentioned the
`trial` case. Also enforced `platform_default` grants must not carry a
`plan_id` (implied by "no `plan_id`" in the task text, made an explicit
invariant). `tests/test_domain.py` covers all listed cases plus the
reserved-tenant extension and double-revoke rejection. `pytest` →
29 passed (whole-repo total, includes T03's tests below).
## Task: Storage layer
```task
id: TEN-WP-0002-T03
status: todo
status: done
priority: high
state_hub_task_id: "ade9374a-ed26-45da-a424-a0d21effe520"
```
@ -103,6 +123,14 @@ Contract — the event bus itself can be a simple in-process list for now.
Done when: unit tests cover create/read/grant/revoke/assign-plan through the
store interface, plus event emission for each mutation.
**Done 2026-07-23:** `store.py`'s `TenantStore` Protocol +
`InMemoryTenantStore` implement create/get/grant/revoke/active_roles/
assign_plan, each emitting a `DomainEvent`. `tests/test_store.py` covers
the full lifecycle, non-exclusive role coexistence (`CUS` + `VEN`
simultaneously, proving ADR-0014's non-exclusivity isn't just a comment),
revoke-keeps-record-but-clears-active-set, and asserts the exact event
sequence for a full mutation chain. `python -m compileall src tests` clean.
## Task: Cache-read API (for key-cape)
```task