T04: deterministic semantic kernel

Alice/Bob/Carol runs end to end, deterministically, replayable from seed.
16 tests pass, no third-party dependencies.

- src/testdriver: intent, provenance, world, actions, drivers, observers,
  oracles, evidence, energy, scenario, runner
- lab/minimal.py: the SUT, exposing the independent observation channel
  required by D-07
- evidence is stratified S1/S2/S3; Runner refuses to attribute S2/S3 to an
  actor; claims are frozen and provenance-checked at construction
- missing evidence yields INCONCLUSIVE, which outranks PASS in the run verdict
- EnergyEvents captured, no scoring (H-005 dormant)

The observation channel records both stored state and an out-of-band
enforcement probe; their disagreement is an invariant and is what detects an
authorization defect that leaves the audit trail intact. A seeded
RevokeIsCosmetic lab fails the run via both the claim and that invariant.

Also closes TD-WP-0001-T02 (stack and commands now exist).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 1629012@bnt-lap001
Assistant-Session: 78d4fb13-8a1e-474b-87a3-9b9261c49a39
This commit is contained in:
tegwick 2026-08-22 23:21:07 +02:00
parent 8da4c5bf7a
commit 04e9573b5a
42 changed files with 1533 additions and 20 deletions

View file

@ -133,6 +133,37 @@ curl -s -X PATCH "http://127.0.0.1:8000/tasks/<task_id>" \
{CREDENTIAL_ROUTING}
<!-- REPO-AGENTS-EXTENSIONS -->
## Stack and commands
Python ≥ 3.11, stdlib only; pytest for the suite. No package manager step is
needed — `pyproject.toml` puts `src/` and the repo root on `pythonpath`.
```bash
python3 -m pytest -q # run everything
python3 -m pytest -q -k oracle # narrow
```
Deliberately boring by decision (`docs/TestDriverResearchPrototype.md`): one
process, one database, one browser engine, one application under test. Novelty
belongs in the verification model, never in the infrastructure. Do not add a
dependency without a stated reason in the workplan.
## Non-negotiables
These are architectural, not stylistic. Breaking one silently defeats the
framework's purpose — see `docs/TestDriverClassificationDesign.md`.
- **Claims and invariants are run inputs.** Never add a code path that lets
adaptation, retry, or a learned trajectory modify them (D-02).
- **S2/S3 evidence is never collected by an actor.** `Runner` enforces this;
do not weaken the check (D-01).
- **Model judgment is confined to S1** — locating controls, proposing paths.
Never verdicts, never claim evaluation (Concept Model § 2.3).
- **Missing evidence yields `INCONCLUSIVE`**, never a default pass or fail.
- **Claims require independent provenance.** `agent-from-implementation` output
is an exploratory hypothesis until a human promotes it (D-06).
<!-- Append repo-specific agent instructions below this marker.
The state-hub template sync preserves content after this line. -->

View file

@ -1,3 +1,41 @@
# test-driver
Agentic framework for integration, end2end, multiuserinteraction, security testing based on usecases.
Agentic framework for integration, end-to-end, multi-user interaction and
security testing, driven by use cases.
Tests mature alongside the software they protect: fluid and agentic while
behaviour is changing, deterministic once it settles. See `INTENT.md` for the
thesis and `SCOPE.md` for boundaries.
**Status:** research prototype. The deterministic kernel runs; agentic
realization, adaptation classification and crystallization are not built yet.
Current work: `workplans/TD-WP-0002-vertical-spike-crystallization.md`.
## Run
```bash
python3 -m pytest -q # the whole suite
python3 -m pytest -q tests/test_reference_scenario.py
```
No third-party dependencies. Python ≥ 3.11, pytest for the suite.
## Layout
```
src/testdriver/ the kernel — intent, world, actions, drivers,
observers, oracles, evidence, runner
lab/ the system under test
scenarios/ reference scenarios
research/ hypotheses, experiments, findings, fitness map
docs/ concept model, improvement loop, milestones, design notes
history/ assessments and completed-work write-ups
```
## Reading order
1. `INTENT.md` — the thesis
2. `docs/TestDriverConceptModel.md` — canonical concept set
3. `docs/TestDriverClassificationDesign.md` — why adaptation cannot normalize a
defect, and where model judgment is and is not permitted
4. `docs/TestDriverInitialMilestones.md` — canonical milestones M0M10

View file

@ -8,14 +8,14 @@
| Kind | ID | Status | Lane | Source |
| --- | --- | --- | --- | --- |
| workplan | TD-WP-0001 | ready | — | workplans/TD-WP-0001-statehub-bootstrap.md |
| workplan | TD-WP-0001 | active | — | workplans/TD-WP-0001-statehub-bootstrap.md |
| workplan | TD-WP-0002 | active | — | workplans/TD-WP-0002-vertical-spike-crystallization.md |
| task | TD-WP-0001-T01 | done | — | workplans/TD-WP-0001-statehub-bootstrap.md |
| task | TD-WP-0001-T02 | wait | — | workplans/TD-WP-0001-statehub-bootstrap.md |
| task | TD-WP-0001-T03 | done | — | workplans/TD-WP-0001-statehub-bootstrap.md |
| task | TD-WP-0002-T01 | done | — | workplans/TD-WP-0002-vertical-spike-crystallization.md |
| task | TD-WP-0002-T02 | done | — | workplans/TD-WP-0002-vertical-spike-crystallization.md |
| task | TD-WP-0002-T03 | todo | — | workplans/TD-WP-0002-vertical-spike-crystallization.md |
| task | TD-WP-0002-T03 | done | — | workplans/TD-WP-0002-vertical-spike-crystallization.md |
| task | TD-WP-0002-T04 | todo | — | workplans/TD-WP-0002-vertical-spike-crystallization.md |
| task | TD-WP-0002-T05 | todo | — | workplans/TD-WP-0002-vertical-spike-crystallization.md |
| task | TD-WP-0002-T06 | todo | — | workplans/TD-WP-0002-vertical-spike-crystallization.md |

1
lab/__init__.py Normal file
View file

@ -0,0 +1 @@
"""The test-driver lab — the system under test."""

Binary file not shown.

Binary file not shown.

191
lab/minimal.py Normal file
View file

@ -0,0 +1,191 @@
"""Minimal lab: the seed of the system under test.
Deliberately small. Users own resources, share them with read or write
permission, and revoke that sharing. Every state change is audited.
Two access paths exist, and the distinction is the whole point:
* the **enforcement path** (`request`) is what an actor uses. It checks
authorization and can therefore be *wrong* that is where a seeded
authorization defect lives.
* the **observation channel** (`ObservationChannel`) reads stored state directly,
without authorization. It is the independent channel required by decision
D-07.
An oracle that consulted only stored state would verify the framework's own
reimplementation of the rules rather than the system's enforcement of them. An
oracle that consulted only the enforcement path would have no way to notice that
enforcement and record disagree. test-driver observes both, and treats
disagreement between them as meaningful in its own right that disagreement is
the precise signature of the M05 authorization defect.
TD-WP-0002-T05 grows this into the full lab with an HTTP API, a browser UI and
the labelled mutation catalogue. It is kept in-process here so that T04 can prove
the kernel without dragging in a web stack.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Literal
Permission = Literal["READ", "WRITE"]
class Denied(Exception):
"""The enforcement path refused the request."""
@dataclass(slots=True)
class AuditRecord:
sequence: int
event: str
actor_id: str
resource_id: str | None = None
subject_id: str | None = None
permission: str | None = None
@dataclass(slots=True)
class MinimalLab:
"""In-process resource-sharing service."""
version: str = "lab-0.1.0-baseline"
users: dict[str, str] = field(default_factory=dict) # user_id -> token
_tokens: dict[str, str] = field(default_factory=dict) # token -> user_id
resources: dict[str, dict[str, Any]] = field(default_factory=dict)
grants: dict[tuple[str, str], Permission] = field(default_factory=dict)
audit: list[AuditRecord] = field(default_factory=list)
_seq: int = 0
# -- setup -----------------------------------------------------------
def add_user(self, user_id: str) -> str:
token = f"tok-{user_id}"
self.users[user_id] = token
self._tokens[token] = user_id
return token
def _audit(self, event: str, actor_id: str, **kw: Any) -> None:
self._seq += 1
self.audit.append(AuditRecord(self._seq, event, actor_id, **kw))
def _whoami(self, token: str) -> str:
if token not in self._tokens:
raise Denied("unknown token")
return self._tokens[token]
# -- enforcement path (what actors use) ------------------------------
def request(self, token: str, op: str, **args: Any) -> Any:
"""The single entry point actors go through. Authorization is enforced here."""
user_id = self._whoami(token)
handler = getattr(self, f"_op_{op}", None)
if handler is None:
raise Denied(f"unknown operation {op!r}")
return handler(user_id, **args)
def _op_create_resource(self, user_id: str, resource_id: str, content: str) -> dict:
self.resources[resource_id] = {"owner": user_id, "content": content}
self._audit("create", user_id, resource_id=resource_id)
return {"resource_id": resource_id}
def _op_read_resource(self, user_id: str, resource_id: str) -> dict:
resource = self.resources.get(resource_id)
if resource is None:
raise Denied("no such resource")
if not self._may_read(user_id, resource_id):
raise Denied("not authorized to read")
return {"resource_id": resource_id, "content": resource["content"]}
def _op_grant(
self, user_id: str, resource_id: str, subject_id: str, permission: Permission
) -> dict:
resource = self.resources.get(resource_id)
if resource is None or resource["owner"] != user_id:
raise Denied("only the owner may grant")
self.grants[(resource_id, subject_id)] = permission
self._audit(
"grant", user_id, resource_id=resource_id,
subject_id=subject_id, permission=permission,
)
return {"granted": permission}
def _op_revoke(self, user_id: str, resource_id: str, subject_id: str) -> dict:
resource = self.resources.get(resource_id)
if resource is None or resource["owner"] != user_id:
raise Denied("only the owner may revoke")
self.grants.pop((resource_id, subject_id), None)
self._audit("revoke", user_id, resource_id=resource_id, subject_id=subject_id)
return {"revoked": True}
def _may_read(self, user_id: str, resource_id: str) -> bool:
"""The authorization rule as the system actually enforces it."""
resource = self.resources.get(resource_id)
if resource is None:
return False
if resource["owner"] == user_id:
return True
return (resource_id, user_id) in self.grants
class ObservationChannel:
"""Independent read access to lab state — decision D-07.
Bypasses authorization deliberately. This is the channel test-driver requires
of any system under test, and the main integration burden the framework
imposes on an adopter.
"""
def __init__(self, lab: MinimalLab) -> None:
self._lab = lab
@property
def version(self) -> str:
return self._lab.version
def state_permission(self, user_id: str, resource_id: str) -> str | None:
"""What the stored record says, independent of any enforcement decision."""
resource = self._lab.resources.get(resource_id)
if resource is None:
return None
if resource["owner"] == user_id:
return "OWNER"
return self._lab.grants.get((resource_id, user_id))
def probe_read(self, user_id: str, resource_id: str) -> bool:
"""Exercise the enforcement path out-of-band and report what it did.
This uses the subject's own credentials, which can look like a violation
of actor isolation but is not: independence means the *actor's report* is
never the evidence. The observer issues its own request and records the
raw outcome. No actor is ever asked whether it succeeded.
"""
token = self._lab.users.get(user_id)
if token is None:
return False
try:
self._lab.request(token, "read_resource", resource_id=resource_id)
except Denied:
return False
return True
def audit_events(self, resource_id: str) -> list[dict[str, Any]]:
return [
{
"sequence": r.sequence,
"event": r.event,
"actor_id": r.actor_id,
"subject_id": r.subject_id,
"permission": r.permission,
}
for r in self._lab.audit
if r.resource_id == resource_id
]
def build_baseline() -> tuple[MinimalLab, dict[str, str]]:
"""Known initial state, so that runs replay from the same starting point."""
lab = MinimalLab()
tokens = {user: lab.add_user(user) for user in ("alice", "bob", "carol")}
return lab, tokens

20
pyproject.toml Normal file
View file

@ -0,0 +1,20 @@
[project]
name = "test-driver"
version = "0.1.0"
description = "Use-case-driven verification framework whose tests mature alongside the software they protect"
requires-python = ">=3.11"
dependencies = []
[project.optional-dependencies]
dev = ["pytest>=7.4"]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src/testdriver"]
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src", "."]

View file

@ -1,6 +1,6 @@
# Concept ↔ Implementation Fitness Map
**Updated:** 2026-08-22 (TD-WP-0002-T03)
**Updated:** 2026-08-22 (TD-WP-0002-T04)
Traces each important concept to the implementation, experiment and evidence that
support it. **Unsupported entries are the point of this map** — a concept with no
@ -13,25 +13,29 @@ Support levels follow `TestDriverImprovementLoop.md` §13:
## Current state
At the time of writing there is **no implementation**, so no concept can honestly
exceed `C1`. The initial classifications in §13 of the Improvement Loop
The deterministic kernel exists (T04) and its guarantees are covered by unit
tests. **Levels have not moved.** A passing unit test is not an experiment: it
shows the code does what its author intended, not that the concept holds under
the mutations it claims to survive. Levels rise when E-001/E-002/E-003 produce
evidence, not before. The implementation column below moves; the level column
does not. The initial classifications in §13 of the Improvement Loop
(`Actor Isolation C2`, `Independent Oracles C2`) are corrected downward here: they
were aspirational, not evidenced.
| Concept | Level | Implementation | Experiment | Evidence | Open question |
|---|---|---|---|---|---|
| `C-use-case` | C1 | — (T04) | — | — | Is a use case expressible without leaking mechanics? |
| `C-actor-isolation` | C1 | — (T04) | E-001 | — | Isolation is asserted by construction; unverified. |
| `C-semantic-action` | C1 | — (T04) | E-001 | — | Does identity survive restructuring better than a recorded sequence? (H-001) |
| `C-oracle-independence` | C1 | — (T04) | E-001, E-003 | — | Independence of components ≠ independence of belief. (H-004) |
| `C-evidence-pack` | C1 | — (T04) | — | — | What is the minimum sufficient for replay? |
| `C-observation-channel` | C1 | — (T05) | — | — | **D-07** — required of every system under test. Adoption cost unknown. |
| `C-use-case` | C1 | `intent.py` | — | — | Is a use case expressible without leaking mechanics? |
| `C-actor-isolation` | C1 | `world.py` | E-001 | — | Isolation is asserted by construction; unverified. |
| `C-semantic-action` | C1 | `actions.py` | E-001 | — | Does identity survive restructuring better than a recorded sequence? (H-001) |
| `C-oracle-independence` | C1 | `runner.py`, `oracles.py` | E-001, E-003 | — | Independence of components ≠ independence of belief. (H-004) |
| `C-evidence-pack` | C1 | `evidence.py` | — | — | What is the minimum sufficient for replay? |
| `C-observation-channel` | C1 | `lab/minimal.py` | — | — | **D-07** — required of every system under test. Adoption cost unknown. |
| `C-adaptation` | C1 | — (T08) | E-001 | — | (H-002) |
| `C-classification` | C1 | — (T08) | E-001, E-003 | — | Decision table is total on paper; unexercised. |
| `C-crystallization` | C1 | — (T09) | E-002 | — | (H-003) |
| `C-intent-provenance` | C1 | — (T04) | E-003 | — | Constrains provenance, not quality. Accepted residual. |
| `C-intent-provenance` | C1 | `provenance.py` | E-003 | — | Constrains provenance, not quality. Accepted residual. |
| `C-lineage` | C0 | — | — | — | Parent pointer only in the spike. |
| `C-energy` | C0 | events only (T04) | — | — | Dormant by decision. (H-005) |
| `C-energy` | C0 | `energy.py`, capture only | — | — | Dormant by decision. (H-005) |
| `C-temperature` | C0 | — | — | — | Deferred. No implementation planned in TD-WP-0002. |
| `C-confidence` | C0 | — | — | — | Deferred. |
| `C-campaign` | C0 | — | — | — | Deferred. |
@ -49,4 +53,7 @@ most at risk of being built because they are easy and satisfying, and never
validated. They are revisited at T10, where the question is not "when do we build
these" but "does the evidence justify keeping them in the model at all".
**Implementation orphans** — none yet; there is no implementation.
**Implementation orphans** — none. Every module in `src/testdriver/` traces to a
concept above. `energy.py` is the one to watch: it exists solely to capture
events for a dormant hypothesis, and if T10 finds no use for the history it
should be removed rather than kept out of sentiment.

Binary file not shown.

View file

@ -0,0 +1,135 @@
"""The reference scenario.
Alice owns resource R. Alice grants Bob READ access. Bob can read R.
Carol cannot read R. Alice revokes Bob's access. Bob can no longer read R.
Written as Python rather than YAML on purpose: claims are predicates over
observations, and a YAML dialect capable of expressing them would be a
programming language with worse tooling. A declarative surface is a later
question, once we know which predicates actually recur.
Every claim here is `Provenance.HUMAN` authored from the use case in INTENT.md,
not derived from watching the lab behave. See D-06.
"""
from __future__ import annotations
from testdriver import (
Actor, Cast, Claim, DirectDriver, Invariant, Oracle, Provenance,
Scenario, SemanticAction, StateObserver, Step, UseCase,
VerificationAsset, Watch, World,
)
from lab.minimal import ObservationChannel, build_baseline
RESOURCE = "R"
API = frozenset({"api"})
# --- claims ---------------------------------------------------------------
# Each reads the independent observation snapshot. None consults an actor.
def _bob_can_read(obs) -> bool:
return obs["probe_read:bob:R"] is True
def _bob_cannot_read(obs) -> bool:
return obs["probe_read:bob:R"] is False
def _carol_cannot_read(obs) -> bool:
return obs["probe_read:carol:R"] is False
def _enforcement_matches_record(obs) -> bool:
"""Enforcement and stored record must agree about every watched subject.
This invariant is what catches an authorization defect that leaves the audit
trail looking correct: the grant is recorded as revoked, yet the enforcement
path still allows the read. Neither observation alone would notice.
"""
for key, permitted in obs.items():
if not key.startswith("probe_read:"):
continue
recorded = obs.get("state_permission:" + key.removeprefix("probe_read:"))
if permitted != (recorded is not None):
return False
return True
def _audit_is_append_only(obs) -> bool:
events = obs["audit:R"]
sequences = [e["sequence"] for e in events]
return sequences == sorted(sequences)
USE_CASE = UseCase(
id="uc-share-resource",
title="Share a resource and revoke the share",
narrative=(
"Alice owns resource R. Alice grants Bob READ access. Bob can read R. "
"Carol cannot read R. Alice revokes Bob's access. "
"Bob can no longer read R."
),
provenance=Provenance.HUMAN,
source_ref="INTENT.md#first-reference-scenario",
claims=(
Claim("c-bob-reads", "Bob can read R after the grant",
Provenance.HUMAN, _bob_can_read, after_step="s2-grant"),
Claim("c-carol-denied", "Carol can never read R",
Provenance.HUMAN, _carol_cannot_read, after_step="s2-grant"),
Claim("c-bob-revoked", "Bob cannot read R after revocation",
Provenance.HUMAN, _bob_cannot_read, after_step="s3-revoke"),
),
invariants=(
Invariant("i-enforcement-matches-record",
"Enforcement and the stored record agree for every subject",
Provenance.HUMAN, _enforcement_matches_record),
Invariant("i-audit-append-only", "The audit trail is append-only",
Provenance.HUMAN, _audit_is_append_only),
),
)
def build(variant: str = "baseline"):
"""Assemble world, driver, observer and asset from a known initial state."""
lab, tokens = build_baseline()
cast = Cast()
for name in ("alice", "bob", "carol"):
cast.add(Actor(id=name, display_name=name.title(),
credentials={"token": tokens[name]}))
world = World(id="w-baseline", sut=lab, sut_version=lab.version,
seed={"users": ["alice", "bob", "carol"], "resource": RESOURCE},
cast=cast)
scenario = Scenario(
id="sc-share-resource",
use_case=USE_CASE,
variant=variant,
watches=(Watch("bob", RESOURCE), Watch("carol", RESOURCE)),
steps=(
Step("s1-create", "alice", SemanticAction(
"create_resource",
{"resource_id": RESOURCE, "content": "the secret"},
permitted_surfaces=API,
postcondition=lambda obs: "audit:R" in obs,
)),
Step("s2-grant", "alice", SemanticAction(
"grant_access",
{"resource_id": RESOURCE, "subject_id": "bob", "permission": "READ"},
permitted_surfaces=API,
postcondition=lambda obs: obs["state_permission:bob:R"] == "READ",
)),
Step("s3-revoke", "alice", SemanticAction(
"revoke_access",
{"resource_id": RESOURCE, "subject_id": "bob"},
permitted_surfaces=API,
postcondition=lambda obs: obs["state_permission:bob:R"] is None,
)),
),
)
driver = DirectDriver(lab, tokens)
observer = StateObserver(ObservationChannel(lab), scenario.watches)
asset = VerificationAsset(id="va-share-resource", scenario=scenario, maturity="T5")
return world, driver, observer, asset, Oracle()

View file

@ -0,0 +1,23 @@
"""test-driver — verification assets that mature alongside the software they protect."""
from .actions import SemanticAction, Surface, SurfaceNotPermitted
from .drivers import DirectDriver, Realization
from .energy import EnergyEvent, EnergyEventType
from .evidence import EvidencePack, Observation, Stratum
from .intent import Claim, Invariant, UseCase
from .observers import StateObserver, Watch
from .oracles import Judgment, Oracle, Verdict, overall
from .provenance import InadmissibleProvenance, Provenance
from .runner import CollectorIndependenceError, Runner, RunResult
from .scenario import Scenario, Step, VerificationAsset
from .world import Actor, ActorIsolationError, Cast, World
__all__ = [
"Actor", "ActorIsolationError", "Cast", "Claim", "CollectorIndependenceError",
"DirectDriver", "EnergyEvent", "EnergyEventType", "EvidencePack",
"InadmissibleProvenance", "Invariant", "Judgment", "Observation", "Oracle",
"Provenance", "Realization", "RunResult", "Runner", "Scenario",
"SemanticAction", "StateObserver", "Step", "Stratum", "Surface",
"SurfaceNotPermitted", "UseCase", "VerificationAsset", "Verdict", "Watch",
"overall",
]

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

57
src/testdriver/actions.py Normal file
View file

@ -0,0 +1,57 @@
"""Semantic actions and the surfaces they may legitimately use.
A semantic action names *what* is being accomplished, never *how*. The `how` is
a driver's business and is expected to change; the `what` is the stable identity
that survives restructuring (H-001).
Every action declares the surfaces it is permitted to use decision D-05. An
actor that achieves `grant_access` through a surface the scenario did not permit
has not recovered from a change; it has performed an unrequested
surface-substitution, which is one of the catalogued security mutations. Without
this check, an agent can route around a broken authorization control and score
as a successful mechanical adaptation.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Mapping
@dataclass(frozen=True, slots=True)
class Surface:
"""An interaction surface: an API, a browser UI, a CLI, a queue."""
id: str
kind: str
description: str = ""
class SurfaceNotPermitted(Exception):
"""An action was realized through a surface the scenario did not permit."""
@dataclass(frozen=True, slots=True)
class SemanticAction:
"""Intent to change or inspect the world, expressed without mechanics.
`postcondition` is evaluated by an independent Observer (S2), never by the
actor that performed the action. An actor never reports its own success.
"""
name: str
args: Mapping[str, Any] = field(default_factory=dict)
permitted_surfaces: frozenset[str] = frozenset()
postcondition: Callable[[Mapping[str, object]], bool] | None = None
def check_surface(self, surface_id: str) -> None:
if self.permitted_surfaces and surface_id not in self.permitted_surfaces:
raise SurfaceNotPermitted(
f"action {self.name!r} was realized through surface "
f"{surface_id!r}, which is not in "
f"{sorted(self.permitted_surfaces)} (D-05)"
)
def describe(self) -> str:
rendered = ", ".join(f"{k}={v!r}" for k, v in sorted(self.args.items()))
return f"{self.name}({rendered})"

76
src/testdriver/drivers.py Normal file
View file

@ -0,0 +1,76 @@
"""Drivers realize semantic actions against a surface.
A driver knows mechanics. It emits S1 evidence describing *how* it acted and
reports which surface it used, but it never reports whether the action was
correct that is S2/S3 and belongs to the observer and the oracle.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Protocol
from .actions import SemanticAction, Surface
from .world import Actor
@dataclass(frozen=True, slots=True)
class Realization:
"""What a driver did, mechanically. Pure S1."""
surface_id: str
mechanics: dict[str, Any]
raised: str | None = None
class Driver(Protocol):
surface: Surface
def realize(self, actor: Actor, action: SemanticAction) -> Realization: ...
class UnsupportedAction(Exception):
"""The driver has no mechanical implementation for this semantic action."""
class DirectDriver:
"""Deterministic driver against the lab's enforcement path.
This is the T5 Deterministic end of the maturity continuum: a fixed mapping
from semantic action to mechanics, with no discovery and no model.
"""
def __init__(self, lab: Any, tokens: dict[str, str]) -> None:
self._lab = lab
self._tokens = tokens
self.surface = Surface(
id="api", kind="http-like", description="lab enforcement path"
)
_MAPPING = {
"create_resource": ("create_resource", ("resource_id", "content")),
"grant_access": ("grant", ("resource_id", "subject_id", "permission")),
"revoke_access": ("revoke", ("resource_id", "subject_id")),
"read_resource": ("read_resource", ("resource_id",)),
}
def realize(self, actor: Actor, action: SemanticAction) -> Realization:
if action.name not in self._MAPPING:
raise UnsupportedAction(action.name)
action.check_surface(self.surface.id)
op, arg_names = self._MAPPING[action.name]
args = {name: action.args[name] for name in arg_names if name in action.args}
token = self._tokens[actor.id]
mechanics: dict[str, Any] = {
"operation": op,
"arguments": args,
"actor": actor.id,
}
try:
result = self._lab.request(token, op, **args)
except Exception as exc: # the SUT refusing is data, not a framework error
return Realization(self.surface.id, mechanics, raised=f"{type(exc).__name__}: {exc}")
mechanics["result"] = result
return Realization(self.surface.id, mechanics)

45
src/testdriver/energy.py Normal file
View file

@ -0,0 +1,45 @@
"""Energy events — capture only.
H-005 is dormant by decision: verification energy is not testable at the current
scale, and a scoring function producing a number nobody can check is worse than
no number. Events are recorded from the first run because history cannot be
reconstructed later; scores always can.
There is deliberately no score() function in this module.
"""
from __future__ import annotations
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any
class EnergyEventType(str, Enum):
DEFECT_DETECTED = "DEFECT_DETECTED"
REGRESSION_CAUGHT = "REGRESSION_CAUGHT"
MECHANICAL_ADAPTATION = "MECHANICAL_ADAPTATION"
SEMANTIC_ADAPTATION = "SEMANTIC_ADAPTATION"
TEST_DEFECT = "TEST_DEFECT"
FALSE_POSITIVE = "FALSE_POSITIVE"
DUPLICATE = "DUPLICATE"
CRYSTALLIZED = "CRYSTALLIZED"
USECASE_DEPRECATED = "USECASE_DEPRECATED"
EXECUTED = "EXECUTED"
@dataclass(frozen=True, slots=True)
class EnergyEvent:
"""Immutable. Energy is derived from event history, never stored as state."""
asset_id: str
run_id: str
event_type: EnergyEventType
detail: dict[str, Any] = field(default_factory=dict)
at: str = field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
def as_dict(self) -> dict[str, Any]:
return {**asdict(self), "event_type": self.event_type.value}

View file

@ -0,0 +1,86 @@
"""Stratified evidence — decision D-01.
S1 Surface how an action was performed (selectors, routes, payloads)
S2 Realization whether it happened, and through which surface
S3 Judgment whether that was correct
Each stratum has a different authority. Models may write S1. Nothing but an
independent Observer writes S2 or S3.
See docs/TestDriverClassificationDesign.md, Part A.
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from enum import Enum
from typing import Any
class Stratum(str, Enum):
SURFACE = "S1"
REALIZATION = "S2"
JUDGMENT = "S3"
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
@dataclass(frozen=True, slots=True)
class Observation:
"""One recorded fact, attributed to a stratum and a collector.
`collector` is never an actor for S2/S3 observations. The runner enforces
this; see runner._assert_collector_independence.
"""
id: str
stratum: Stratum
collector: str
step_id: str | None
kind: str
data: dict[str, Any]
at: str = field(default_factory=_now)
@dataclass(slots=True)
class EvidencePack:
"""Everything retained from one run.
The pack must be sufficient to replay the run and to diagnose a finding
without the original process. An assertion that cannot be supported from the
pack is an EVIDENCE_FAILURE, not a defect in the system under test.
"""
run_id: str
scenario_id: str
use_case_id: str
sut_version: str
started_at: str = field(default_factory=_now)
finished_at: str | None = None
observations: list[Observation] = field(default_factory=list)
verdicts: list[dict[str, Any]] = field(default_factory=list)
energy_events: list[dict[str, Any]] = field(default_factory=list)
provenance_index: dict[str, str] = field(default_factory=dict)
def record(self, observation: Observation) -> None:
self.observations.append(observation)
def of_stratum(self, stratum: Stratum) -> list[Observation]:
return [o for o in self.observations if o.stratum is stratum]
def latest(self, kind: str) -> Observation | None:
for observation in reversed(self.observations):
if observation.kind == kind:
return observation
return None
def to_json(self) -> str:
payload = asdict(self)
payload["observations"] = [
{**asdict(o), "stratum": o.stratum.value} for o in self.observations
]
return json.dumps(payload, indent=2, sort_keys=True, default=str)

63
src/testdriver/intent.py Normal file
View file

@ -0,0 +1,63 @@
"""The intent layer: what is supposed to be true.
Claims and invariants are *inputs* to a run and are frozen decision D-02.
There is deliberately no code path by which adaptation, retry, or a learned
trajectory can modify them. That absence is what makes False Adaptation Rate = 0
an architectural property rather than a tuning target.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Callable, Mapping
from .provenance import Provenance, require_admissible
# A predicate over the independent observations gathered during a run.
# It receives the observation mapping and returns True when satisfied.
Predicate = Callable[[Mapping[str, object]], bool]
@dataclass(frozen=True, slots=True)
class Claim:
"""A statement that must hold at a specific point in a scenario."""
id: str
text: str
provenance: Provenance
predicate: Predicate
after_step: str
source_ref: str | None = None
def __post_init__(self) -> None:
require_admissible(self.provenance, f"Claim {self.id!r}")
@dataclass(frozen=True, slots=True)
class Invariant:
"""A statement that must hold after *every* step, not merely at one point."""
id: str
text: str
provenance: Provenance
predicate: Predicate
source_ref: str | None = None
def __post_init__(self) -> None:
require_admissible(self.provenance, f"Invariant {self.id!r}")
@dataclass(frozen=True, slots=True)
class UseCase:
"""Purposeful behaviour, described independently of mechanics."""
id: str
title: str
narrative: str
provenance: Provenance
claims: tuple[Claim, ...] = field(default_factory=tuple)
invariants: tuple[Invariant, ...] = field(default_factory=tuple)
source_ref: str | None = None
def __post_init__(self) -> None:
require_admissible(self.provenance, f"UseCase {self.id!r}")

View file

@ -0,0 +1,47 @@
"""Observers gather evidence independently of the actors.
An observer never asks an actor what happened. It reads the system directly
through the independent observation channel required by decision D-07, and
records both what the stored record says and what the enforcement path actually
does. Disagreement between those two is meaningful in its own right: it is the
signature of an authorization defect, where the audit trail says one thing and
enforcement does another.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Sequence
@dataclass(frozen=True, slots=True)
class Watch:
"""A (subject, resource) pair the scenario wants observed after every step."""
subject_id: str
resource_id: str
@dataclass(slots=True)
class StateObserver:
"""Collects the S3 snapshot: what is true of the domain right now."""
channel: Any
watches: Sequence[Watch] = field(default_factory=tuple)
name: str = "state-observer"
def snapshot(self) -> dict[str, Any]:
out: dict[str, Any] = {}
resources = set()
for watch in self.watches:
key = f"{watch.subject_id}:{watch.resource_id}"
out[f"probe_read:{key}"] = self.channel.probe_read(
watch.subject_id, watch.resource_id
)
out[f"state_permission:{key}"] = self.channel.state_permission(
watch.subject_id, watch.resource_id
)
resources.add(watch.resource_id)
for resource_id in sorted(resources):
out[f"audit:{resource_id}"] = self.channel.audit_events(resource_id)
return out

106
src/testdriver/oracles.py Normal file
View file

@ -0,0 +1,106 @@
"""Oracles evaluate claims and invariants and produce verdicts.
An oracle reads only the independent observation snapshot. It has no access to
the actor, to the driver, or to what either of them believes happened.
`INCONCLUSIVE` is a first-class outcome, not a failure mode of the framework. An
oracle that cannot support a judgment from the retained evidence must say so
rather than defaulting either way silently defaulting to PASS hides defects,
and silently defaulting to FAIL trains people to ignore results.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Mapping
from .intent import Claim, Invariant
class Verdict(str, Enum):
PASS = "PASS"
FAIL = "FAIL"
SUSPICIOUS = "SUSPICIOUS"
INCONCLUSIVE = "INCONCLUSIVE"
@dataclass(frozen=True, slots=True)
class Judgment:
assertion_id: str
text: str
verdict: Verdict
step_id: str | None
detail: dict[str, Any] = field(default_factory=dict)
def as_dict(self) -> dict[str, Any]:
return {
"assertion_id": self.assertion_id,
"text": self.text,
"verdict": self.verdict.value,
"step_id": self.step_id,
"detail": self.detail,
}
class Oracle:
"""Deterministic evaluation of one assertion against an observation snapshot."""
def judge(
self,
assertion: Claim | Invariant,
snapshot: Mapping[str, Any],
step_id: str | None,
) -> Judgment:
if not snapshot:
return Judgment(
assertion.id,
assertion.text,
Verdict.INCONCLUSIVE,
step_id,
{"reason": "no observations were collected"},
)
try:
satisfied = assertion.predicate(snapshot)
except KeyError as missing:
# The evidence needed to judge this assertion was not collected.
# That is an evidence failure, never a pass and never a fail.
return Judgment(
assertion.id,
assertion.text,
Verdict.INCONCLUSIVE,
step_id,
{"reason": f"required observation {missing} missing from snapshot"},
)
except Exception as exc:
return Judgment(
assertion.id,
assertion.text,
Verdict.INCONCLUSIVE,
step_id,
{"reason": f"predicate raised {type(exc).__name__}: {exc}"},
)
return Judgment(
assertion.id,
assertion.text,
Verdict.PASS if satisfied else Verdict.FAIL,
step_id,
)
def overall(judgments: list[Judgment]) -> Verdict:
"""Aggregate run verdict.
FAIL dominates; INCONCLUSIVE outranks PASS. A run containing an unjudgeable
assertion has not passed, whatever else it did.
"""
verdicts = {j.verdict for j in judgments}
if not judgments:
return Verdict.INCONCLUSIVE
if Verdict.FAIL in verdicts:
return Verdict.FAIL
if Verdict.INCONCLUSIVE in verdicts:
return Verdict.INCONCLUSIVE
if Verdict.SUSPICIOUS in verdicts:
return Verdict.SUSPICIOUS
return Verdict.PASS

View file

@ -0,0 +1,46 @@
"""Claim provenance — decision D-06.
A claim may only be authored by a source causally independent of the
implementation it constrains. Without this rule the framework's guarantee
reduces to "the implementation agrees with itself", which is exactly the failure
the project exists to prevent.
See docs/TestDriverClassificationDesign.md, Part B.
"""
from __future__ import annotations
from enum import Enum
class Provenance(str, Enum):
"""Where an intent artifact came from."""
HUMAN = "human"
SPEC = "spec"
AGENT_FROM_SPEC = "agent-from-spec"
AGENT_FROM_IMPLEMENTATION = "agent-from-implementation"
@property
def admissible_as_claim(self) -> bool:
"""Whether this provenance may back an assertion that can produce FAIL.
`agent-from-implementation` is not forbidden as an activity it is
genuinely useful for T0 exploration. It is forbidden as a *claim*.
Such output enters as an exploratory hypothesis and requires an explicit
human acceptance event before it can constrain the system.
"""
return self is not Provenance.AGENT_FROM_IMPLEMENTATION
class InadmissibleProvenance(Exception):
"""Raised when implementation-derived intent is used as a claim."""
def require_admissible(provenance: Provenance, what: str) -> None:
if not provenance.admissible_as_claim:
raise InadmissibleProvenance(
f"{what} has provenance {provenance.value!r}, which is derived from the "
"implementation it would constrain. Promote it through an explicit human "
"acceptance event before using it as a claim (D-06)."
)

190
src/testdriver/runner.py Normal file
View file

@ -0,0 +1,190 @@
"""The orchestrator: executes a scenario and assembles its evidence.
The runner is the only component that sees everything. Actors see their own
credentials and memory; drivers see mechanics; observers see the system; oracles
see the observation snapshot. Keeping those views separate is what makes the
independence claim structural rather than procedural.
"""
from __future__ import annotations
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
from .actions import SurfaceNotPermitted
from .drivers import Driver
from .energy import EnergyEvent, EnergyEventType
from .evidence import EvidencePack, Observation, Stratum
from .observers import StateObserver
from .oracles import Judgment, Oracle, Verdict, overall
from .scenario import Scenario, VerificationAsset
from .world import World
class CollectorIndependenceError(Exception):
"""An actor was about to be recorded as the collector of S2/S3 evidence."""
@dataclass(slots=True)
class RunResult:
run_id: str
verdict: Verdict
judgments: list[Judgment]
evidence: EvidencePack
def judgment(self, assertion_id: str) -> Judgment:
for j in self.judgments:
if j.assertion_id == assertion_id:
return j
raise KeyError(assertion_id)
class Runner:
def __init__(
self,
world: World,
driver: Driver,
observer: StateObserver,
oracle: Oracle | None = None,
) -> None:
self._world = world
self._driver = driver
self._observer = observer
self._oracle = oracle or Oracle()
# -- independence guards ---------------------------------------------
def _assert_collector_independence(self, stratum: Stratum, collector: str) -> None:
"""S2 and S3 evidence may never be attributed to an actor.
This is the check that makes oracle independence falsifiable rather than
merely asserted: wiring an actor in as an observer fails loudly here.
"""
if stratum is Stratum.SURFACE:
return
if collector in self._world.cast.actors:
raise CollectorIndependenceError(
f"{stratum.value} evidence cannot be collected by actor "
f"{collector!r}; actors do not judge their own outcomes"
)
def _record(
self,
pack: EvidencePack,
stratum: Stratum,
collector: str,
kind: str,
data: dict[str, Any],
step_id: str | None,
) -> None:
self._assert_collector_independence(stratum, collector)
pack.record(
Observation(
id=f"obs-{len(pack.observations) + 1:04d}",
stratum=stratum,
collector=collector,
step_id=step_id,
kind=kind,
data=data,
)
)
# -- execution --------------------------------------------------------
def run(self, asset: VerificationAsset) -> RunResult:
scenario: Scenario = asset.scenario
run_id = f"run-{uuid.uuid4().hex[:12]}"
pack = EvidencePack(
run_id=run_id,
scenario_id=scenario.id,
use_case_id=scenario.use_case.id,
sut_version=self._world.sut_version,
)
pack.provenance_index = {
scenario.use_case.id: scenario.use_case.provenance.value,
**{c.id: c.provenance.value for c in scenario.use_case.claims},
**{i.id: i.provenance.value for i in scenario.use_case.invariants},
}
pack.energy_events.append(
EnergyEvent(asset.id, run_id, EnergyEventType.EXECUTED).as_dict()
)
judgments: list[Judgment] = []
claims_by_step: dict[str, list] = {}
for claim in scenario.use_case.claims:
claims_by_step.setdefault(claim.after_step, []).append(claim)
for step in scenario.steps:
actor = self._world.cast[step.actor_id]
# --- S1: how it was done -------------------------------------
try:
realization = self._driver.realize(actor, step.action)
except SurfaceNotPermitted as exc:
# D-05: routing around a control is a finding, not a recovery.
self._record(
pack, Stratum.REALIZATION, self._observer.name,
"surface_violation",
{"step": step.id, "action": step.action.describe(), "error": str(exc)},
step.id,
)
break
self._record(
pack, Stratum.SURFACE, actor.id, "realization",
{
"action": step.action.describe(),
"surface": realization.surface_id,
"mechanics": realization.mechanics,
"raised": realization.raised,
},
step.id,
)
# --- S3: what is now true ------------------------------------
snapshot = self._observer.snapshot()
self._record(
pack, Stratum.JUDGMENT, self._observer.name,
"state_snapshot", dict(snapshot), step.id,
)
# --- S2: did the action actually take effect ------------------
refused = realization.raised is not None
postcondition_met: bool | None = None
if step.action.postcondition is not None:
try:
postcondition_met = step.action.postcondition(snapshot)
except KeyError:
postcondition_met = None
self._record(
pack, Stratum.REALIZATION, self._observer.name, "realization_check",
{
"step": step.id,
"action": step.action.name,
"surface_used": realization.surface_id,
"refused_by_sut": refused,
"refusal_expected": step.expect_refusal,
"postcondition_met": postcondition_met,
},
step.id,
)
# --- invariants after every step -----------------------------
for invariant in scenario.use_case.invariants:
judgments.append(self._oracle.judge(invariant, snapshot, step.id))
# --- claims attached to this step ----------------------------
for claim in claims_by_step.get(step.id, ()):
judgments.append(self._oracle.judge(claim, snapshot, step.id))
pack.verdicts = [j.as_dict() for j in judgments]
pack.finished_at = datetime.now(timezone.utc).isoformat()
result_verdict = overall(judgments)
if result_verdict is Verdict.FAIL:
pack.energy_events.append(
EnergyEvent(asset.id, run_id, EnergyEventType.DEFECT_DETECTED).as_dict()
)
return RunResult(run_id, result_verdict, judgments, pack)

View file

@ -0,0 +1,47 @@
"""A scenario binds a use case to concrete actors, a world and a schedule."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Sequence
from .actions import SemanticAction
from .intent import UseCase
from .observers import Watch
@dataclass(frozen=True, slots=True)
class Step:
"""One scheduled semantic action, attributed to one actor."""
id: str
actor_id: str
action: SemanticAction
expect_refusal: bool = False
@dataclass(frozen=True, slots=True)
class Scenario:
"""UseCase + Actors + World + Schedule + Surfaces + Variant."""
id: str
use_case: UseCase
steps: tuple[Step, ...]
watches: tuple[Watch, ...] = field(default_factory=tuple)
variant: str = "baseline"
@dataclass(slots=True)
class VerificationAsset:
"""A test as a durable thing with identity, maturity and lineage.
Maturity is the T0..T5 continuum. The kernel produces T5 assets: fully
deterministic, no model involvement. Agentic assets (T1) arrive in T07, and
crystallization walks an asset from T1 toward T5 in T09.
"""
id: str
scenario: Scenario
maturity: str = "T5"
parent_id: str | None = None
adaptation_history: list[dict] = field(default_factory=list)

79
src/testdriver/world.py Normal file
View file

@ -0,0 +1,79 @@
"""The situation layer: who is acting, and in what state of the world.
Actor isolation is structural. An Actor holds its own credentials and private
memory and has no reference to the Cast or to any sibling. The orchestrator may
know the whole world; actors must not learn anything merely because they happen
to be executed by the same process.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Iterator
class ActorIsolationError(Exception):
"""Raised when one actor is offered another actor's private state."""
@dataclass(slots=True)
class Actor:
"""An independent execution entity.
Deliberately holds no back-reference to the Cast or the World. An actor that
can enumerate its siblings can leak knowledge it was never given, and no
later check can reliably detect that it did.
"""
id: str
display_name: str
credentials: dict[str, str] = field(default_factory=dict)
session: dict[str, Any] = field(default_factory=dict)
_memory: dict[str, Any] = field(default_factory=dict, repr=False)
def remember(self, key: str, value: Any) -> None:
self._memory[key] = value
def recall(self, key: str, default: Any = None) -> Any:
return self._memory.get(key, default)
def known_keys(self) -> tuple[str, ...]:
return tuple(sorted(self._memory))
@dataclass(slots=True)
class Cast:
"""The set of actors in a scenario. Known to the orchestrator, not to actors."""
actors: dict[str, Actor] = field(default_factory=dict)
def add(self, actor: Actor) -> Actor:
if actor.id in self.actors:
raise ValueError(f"duplicate actor id {actor.id!r}")
self.actors[actor.id] = actor
return actor
def __getitem__(self, actor_id: str) -> Actor:
return self.actors[actor_id]
def __iter__(self) -> Iterator[Actor]:
return iter(self.actors.values())
def __len__(self) -> int:
return len(self.actors)
@dataclass(slots=True)
class World:
"""Initial state plus the handle to the system under test.
`seed` is everything needed to rebuild the initial state, so that a run can
be replayed from a known starting point rather than from wherever the
previous run happened to leave things.
"""
id: str
sut: Any
sut_version: str
seed: dict[str, Any] = field(default_factory=dict)
cast: Cast = field(default_factory=Cast)

View file

@ -0,0 +1,134 @@
"""The kernel must enforce the properties the T02 design claims for it.
These are unit-level checks on the kernel's construction. The behavioural
self-verification suite (actor isolation, oracle independence, evidence
reproducibility, intent independence) is TD-WP-0002-T06 and lives separately.
"""
from __future__ import annotations
import dataclasses
import pytest
from testdriver import (
Claim, InadmissibleProvenance, Invariant, Oracle, Provenance, Runner,
SemanticAction, StateObserver, SurfaceNotPermitted, Stratum, Verdict,
)
from lab.minimal import Denied, MinimalLab, ObservationChannel, build_baseline
from scenarios.alice_bob_carol import build
def test_implementation_derived_claims_are_rejected():
"""D-06: a claim derived from watching the system cannot constrain it."""
with pytest.raises(InadmissibleProvenance) as exc:
Claim(
"c-observed", "whatever the system currently does",
Provenance.AGENT_FROM_IMPLEMENTATION,
lambda obs: True, after_step="s1",
)
assert "derived from the implementation" in str(exc.value)
def test_spec_and_human_provenance_are_admissible():
for provenance in (Provenance.HUMAN, Provenance.SPEC, Provenance.AGENT_FROM_SPEC):
Invariant("i-ok", "fine", provenance, lambda obs: True)
def test_claims_are_frozen():
"""D-02: claims are run inputs. No adaptation path may rewrite them."""
claim = Claim("c", "text", Provenance.HUMAN, lambda obs: True, after_step="s1")
with pytest.raises(dataclasses.FrozenInstanceError):
claim.text = "something the implementation would prefer"
def test_surface_substitution_is_refused():
"""D-05: achieving an action through an unpermitted surface is not recovery."""
action = SemanticAction(
"grant_access", {"resource_id": "R"}, permitted_surfaces=frozenset({"browser"})
)
with pytest.raises(SurfaceNotPermitted):
action.check_surface("api")
def test_missing_evidence_yields_inconclusive_not_pass():
"""An unjudgeable assertion must never default either way."""
claim = Claim(
"c-needs-missing", "depends on evidence nobody collected",
Provenance.HUMAN, lambda obs: obs["probe_read:dave:R"], after_step="s1",
)
judgment = Oracle().judge(claim, {"probe_read:bob:R": True}, "s1")
assert judgment.verdict is Verdict.INCONCLUSIVE
assert "missing from snapshot" in judgment.detail["reason"]
def test_inconclusive_outranks_pass_in_the_run_verdict():
from testdriver.oracles import Judgment, overall
assert overall([
Judgment("a", "", Verdict.PASS, None),
Judgment("b", "", Verdict.INCONCLUSIVE, None),
]) is Verdict.INCONCLUSIVE
def test_actors_cannot_be_recorded_as_judgment_collectors():
"""Wiring an actor in as an observer must fail loudly, not silently pass."""
world, driver, observer, asset, oracle = build()
observer.name = "alice" # an actor pretending to be the observer
with pytest.raises(Exception) as exc:
Runner(world, driver, observer, oracle).run(asset)
assert "do not judge their own outcomes" in str(exc.value)
# --- preview of the M05 authorization defect (built properly in T05) --------
class RevokeIsCosmetic(MinimalLab):
"""Revocation updates the record and the audit trail but not enforcement.
This is the shape of a real authorization defect: everything an operator
would look at says access was removed, and access was not removed.
"""
def _may_read(self, user_id: str, resource_id: str) -> bool:
resource = self.resources.get(resource_id)
if resource is None:
return False
if resource["owner"] == user_id:
return True
return any(
r.event == "grant" and r.subject_id == user_id and r.resource_id == resource_id
for r in self.audit
)
def test_seeded_authorization_defect_fails_the_run():
"""The kernel must report FAIL, not adapt, when revocation does not revoke."""
from testdriver import Actor, Cast, DirectDriver, World
from scenarios.alice_bob_carol import USE_CASE
from scenarios import alice_bob_carol as ref
lab = RevokeIsCosmetic(version="lab-0.1.0-M05-auth-defect")
tokens = {u: lab.add_user(u) for u in ("alice", "bob", "carol")}
cast = Cast()
for name in tokens:
cast.add(Actor(name, name.title(), credentials={"token": tokens[name]}))
_, _, _, baseline_asset, _ = ref.build()
world = World("w-defect", lab, lab.version, cast=cast)
driver = DirectDriver(lab, tokens)
observer = StateObserver(ObservationChannel(lab), baseline_asset.scenario.watches)
result = Runner(world, driver, observer, Oracle()).run(baseline_asset)
assert result.verdict is Verdict.FAIL
assert result.judgment("c-bob-revoked").verdict is Verdict.FAIL
# The claim set is untouched by the failure — there is no path to adapt it.
assert USE_CASE.claims[2].text == "Bob cannot read R after revocation"
def test_defect_run_emits_an_energy_event():
"""Energy events are captured; no score is computed (H-005 is dormant)."""
import testdriver.energy as energy
assert not hasattr(energy, "score")

View file

@ -0,0 +1,70 @@
"""The reference scenario must run deterministically and be replayable."""
from __future__ import annotations
import json
import pytest
from testdriver import Runner, Stratum, Verdict
from scenarios.alice_bob_carol import build
def run_once(variant: str = "baseline"):
world, driver, observer, asset, oracle = build(variant)
return Runner(world, driver, observer, oracle).run(asset), world
def test_reference_scenario_passes():
result, _ = run_once()
assert result.verdict is Verdict.PASS, [
j.as_dict() for j in result.judgments if j.verdict is not Verdict.PASS
]
def test_every_claim_is_judged():
result, _ = run_once()
judged = {j.assertion_id for j in result.judgments}
assert {"c-bob-reads", "c-carol-denied", "c-bob-revoked"} <= judged
def test_invariants_are_evaluated_after_every_step():
result, _ = run_once()
per_step = [j for j in result.judgments if j.assertion_id == "i-audit-append-only"]
assert len(per_step) == 3
def test_run_is_replayable_from_known_initial_state():
"""Two runs from the same seed produce identical judgments."""
first, _ = run_once()
second, _ = run_once()
assert [(j.assertion_id, j.verdict) for j in first.judgments] == [
(j.assertion_id, j.verdict) for j in second.judgments
]
assert first.run_id != second.run_id
def test_evidence_is_stratified_and_serializable():
result, _ = run_once()
pack = result.evidence
assert pack.of_stratum(Stratum.SURFACE)
assert pack.of_stratum(Stratum.REALIZATION)
assert pack.of_stratum(Stratum.JUDGMENT)
parsed = json.loads(pack.to_json())
assert parsed["run_id"] == result.run_id
assert parsed["sut_version"] == "lab-0.1.0-baseline"
def test_evidence_records_claim_provenance():
"""A verdict must be auditable for the independence of the claim behind it."""
result, _ = run_once()
assert result.evidence.provenance_index["c-bob-revoked"] == "human"
def test_actors_hold_isolated_credentials_and_memory():
_, world = run_once()
alice, bob = world.cast["alice"], world.cast["bob"]
assert alice.credentials["token"] != bob.credentials["token"]
alice.remember("secret", "only alice knows this")
assert bob.recall("secret") is None
assert bob.known_keys() == ()

View file

@ -32,7 +32,7 @@ Replace generated placeholders with repo-specific facts where needed.
```task
id: TD-WP-0001-T02
status: wait
status: done
priority: high
state_hub_task_id: "39da4237-aaa3-5b4f-93c4-9cbf794f0850"
```
@ -58,8 +58,9 @@ checkout:
statehub fix-consistency
```
Blocked until the stack exists: no code, no dependency manifest and no test
runner are present yet. Unblocks with TD-WP-0002-T04 (deterministic semantic
kernel), which introduces the first Python package and pytest configuration.
**Done 2026-08-22.** Unblocked by TD-WP-0002-T04. `pyproject.toml` added;
`python3 -m pytest -q` is the whole workflow — stdlib only, no install step.
Commands and the architectural non-negotiables are recorded in `AGENTS.md`
under the repo-extensions marker, and in `README.md`.
Seeded workplan: `workplans/TD-WP-0002-vertical-spike-crystallization.md`.

View file

@ -176,7 +176,7 @@ Three things worth carrying forward:
```task
id: TD-WP-0002-T04
status: todo
status: done
priority: high
state_hub_task_id: "ffbcc9e8-c1bd-5c50-b62d-495ad9e135a4"
```
@ -198,6 +198,26 @@ structured Evidence Pack; the scenario replays from known initial state.
Emit raw `EnergyEvent` records from this point onward. Implement no scoring.
**Done 2026-08-22.** `src/testdriver/` (11 modules), `lab/minimal.py`,
`scenarios/alice_bob_carol.py`, 16 passing tests. The reference scenario runs
end to end and replays identically from the same seed; evidence comes out
stratified 3/3/3 across S1/S2/S3.
Three things that came out of building it rather than designing it:
- **The observation channel needs two probes, not one.** Reading stored state
alone verifies test-driver's reimplementation of the rules rather than the
system's enforcement of them; probing enforcement alone cannot notice that
record and enforcement disagree. The lab exposes both, and their disagreement
is now an invariant (`i-enforcement-matches-record`). That invariant is what
catches an authorization defect which leaves the audit trail looking correct.
- **A seeded `RevokeIsCosmetic` lab already fails the run** — both the claim and
the independent invariant fire, and the claim set is provably untouched. Early
evidence for H-004, though not yet the experiment.
- **Scenarios are Python, not YAML.** Claims are predicates over observations; a
YAML dialect able to express them would be a programming language with worse
tooling. Revisit once we know which predicates actually recur.
## Test-driver lab with labelled ground truth
```task