diff --git a/AGENTS.md b/AGENTS.md index a3f37ec..f6d4620 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -133,6 +133,37 @@ curl -s -X PATCH "http://127.0.0.1:8000/tasks/" \ {CREDENTIAL_ROUTING} + +## 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). + diff --git a/README.md b/README.md index 4d1e923..8a3d0d8 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,41 @@ # test-driver -Agentic framework for integration, end2end, multiuserinteraction, security testing based on usecases. \ No newline at end of file +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 M0–M10 diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index f0a9173..eeeaba5 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -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 | diff --git a/lab/__init__.py b/lab/__init__.py new file mode 100644 index 0000000..0e1aa10 --- /dev/null +++ b/lab/__init__.py @@ -0,0 +1 @@ +"""The test-driver lab — the system under test.""" diff --git a/lab/__pycache__/__init__.cpython-312.pyc b/lab/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..930c9db Binary files /dev/null and b/lab/__pycache__/__init__.cpython-312.pyc differ diff --git a/lab/__pycache__/minimal.cpython-312.pyc b/lab/__pycache__/minimal.cpython-312.pyc new file mode 100644 index 0000000..f1dabc8 Binary files /dev/null and b/lab/__pycache__/minimal.cpython-312.pyc differ diff --git a/lab/minimal.py b/lab/minimal.py new file mode 100644 index 0000000..9864964 --- /dev/null +++ b/lab/minimal.py @@ -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 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..c4b9c17 --- /dev/null +++ b/pyproject.toml @@ -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", "."] diff --git a/research/concepts/fitness-map.md b/research/concepts/fitness-map.md index f6c7630..ea72475 100644 --- a/research/concepts/fitness-map.md +++ b/research/concepts/fitness-map.md @@ -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. diff --git a/scenarios/__pycache__/alice_bob_carol.cpython-312.pyc b/scenarios/__pycache__/alice_bob_carol.cpython-312.pyc new file mode 100644 index 0000000..a6b11c1 Binary files /dev/null and b/scenarios/__pycache__/alice_bob_carol.cpython-312.pyc differ diff --git a/scenarios/alice_bob_carol.py b/scenarios/alice_bob_carol.py new file mode 100644 index 0000000..4d478d6 --- /dev/null +++ b/scenarios/alice_bob_carol.py @@ -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() diff --git a/src/testdriver/__init__.py b/src/testdriver/__init__.py new file mode 100644 index 0000000..d56c3ca --- /dev/null +++ b/src/testdriver/__init__.py @@ -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", +] diff --git a/src/testdriver/__pycache__/__init__.cpython-312.pyc b/src/testdriver/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..216faa6 Binary files /dev/null and b/src/testdriver/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/testdriver/__pycache__/actions.cpython-312.pyc b/src/testdriver/__pycache__/actions.cpython-312.pyc new file mode 100644 index 0000000..7b9fe8e Binary files /dev/null and b/src/testdriver/__pycache__/actions.cpython-312.pyc differ diff --git a/src/testdriver/__pycache__/drivers.cpython-312.pyc b/src/testdriver/__pycache__/drivers.cpython-312.pyc new file mode 100644 index 0000000..6367df1 Binary files /dev/null and b/src/testdriver/__pycache__/drivers.cpython-312.pyc differ diff --git a/src/testdriver/__pycache__/energy.cpython-312.pyc b/src/testdriver/__pycache__/energy.cpython-312.pyc new file mode 100644 index 0000000..f2f2a7f Binary files /dev/null and b/src/testdriver/__pycache__/energy.cpython-312.pyc differ diff --git a/src/testdriver/__pycache__/evidence.cpython-312.pyc b/src/testdriver/__pycache__/evidence.cpython-312.pyc new file mode 100644 index 0000000..887c8f1 Binary files /dev/null and b/src/testdriver/__pycache__/evidence.cpython-312.pyc differ diff --git a/src/testdriver/__pycache__/intent.cpython-312.pyc b/src/testdriver/__pycache__/intent.cpython-312.pyc new file mode 100644 index 0000000..7868e9a Binary files /dev/null and b/src/testdriver/__pycache__/intent.cpython-312.pyc differ diff --git a/src/testdriver/__pycache__/observers.cpython-312.pyc b/src/testdriver/__pycache__/observers.cpython-312.pyc new file mode 100644 index 0000000..193b6e8 Binary files /dev/null and b/src/testdriver/__pycache__/observers.cpython-312.pyc differ diff --git a/src/testdriver/__pycache__/oracles.cpython-312.pyc b/src/testdriver/__pycache__/oracles.cpython-312.pyc new file mode 100644 index 0000000..7980776 Binary files /dev/null and b/src/testdriver/__pycache__/oracles.cpython-312.pyc differ diff --git a/src/testdriver/__pycache__/provenance.cpython-312.pyc b/src/testdriver/__pycache__/provenance.cpython-312.pyc new file mode 100644 index 0000000..71c7136 Binary files /dev/null and b/src/testdriver/__pycache__/provenance.cpython-312.pyc differ diff --git a/src/testdriver/__pycache__/runner.cpython-312.pyc b/src/testdriver/__pycache__/runner.cpython-312.pyc new file mode 100644 index 0000000..09410a6 Binary files /dev/null and b/src/testdriver/__pycache__/runner.cpython-312.pyc differ diff --git a/src/testdriver/__pycache__/scenario.cpython-312.pyc b/src/testdriver/__pycache__/scenario.cpython-312.pyc new file mode 100644 index 0000000..8bf7e17 Binary files /dev/null and b/src/testdriver/__pycache__/scenario.cpython-312.pyc differ diff --git a/src/testdriver/__pycache__/world.cpython-312.pyc b/src/testdriver/__pycache__/world.cpython-312.pyc new file mode 100644 index 0000000..6762964 Binary files /dev/null and b/src/testdriver/__pycache__/world.cpython-312.pyc differ diff --git a/src/testdriver/actions.py b/src/testdriver/actions.py new file mode 100644 index 0000000..92dd09d --- /dev/null +++ b/src/testdriver/actions.py @@ -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})" diff --git a/src/testdriver/drivers.py b/src/testdriver/drivers.py new file mode 100644 index 0000000..bb8cdc8 --- /dev/null +++ b/src/testdriver/drivers.py @@ -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) diff --git a/src/testdriver/energy.py b/src/testdriver/energy.py new file mode 100644 index 0000000..23b94f2 --- /dev/null +++ b/src/testdriver/energy.py @@ -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} diff --git a/src/testdriver/evidence.py b/src/testdriver/evidence.py new file mode 100644 index 0000000..01a65a8 --- /dev/null +++ b/src/testdriver/evidence.py @@ -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) diff --git a/src/testdriver/intent.py b/src/testdriver/intent.py new file mode 100644 index 0000000..89fd427 --- /dev/null +++ b/src/testdriver/intent.py @@ -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}") diff --git a/src/testdriver/observers.py b/src/testdriver/observers.py new file mode 100644 index 0000000..40de7bb --- /dev/null +++ b/src/testdriver/observers.py @@ -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 diff --git a/src/testdriver/oracles.py b/src/testdriver/oracles.py new file mode 100644 index 0000000..705b1ae --- /dev/null +++ b/src/testdriver/oracles.py @@ -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 diff --git a/src/testdriver/provenance.py b/src/testdriver/provenance.py new file mode 100644 index 0000000..8df49ce --- /dev/null +++ b/src/testdriver/provenance.py @@ -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)." + ) diff --git a/src/testdriver/runner.py b/src/testdriver/runner.py new file mode 100644 index 0000000..08fdff4 --- /dev/null +++ b/src/testdriver/runner.py @@ -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) diff --git a/src/testdriver/scenario.py b/src/testdriver/scenario.py new file mode 100644 index 0000000..a41de6e --- /dev/null +++ b/src/testdriver/scenario.py @@ -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) diff --git a/src/testdriver/world.py b/src/testdriver/world.py new file mode 100644 index 0000000..04f2341 --- /dev/null +++ b/src/testdriver/world.py @@ -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) diff --git a/tests/__pycache__/test_kernel_guarantees.cpython-312-pytest-7.4.4.pyc b/tests/__pycache__/test_kernel_guarantees.cpython-312-pytest-7.4.4.pyc new file mode 100644 index 0000000..cfc3f09 Binary files /dev/null and b/tests/__pycache__/test_kernel_guarantees.cpython-312-pytest-7.4.4.pyc differ diff --git a/tests/__pycache__/test_kernel_guarantees.cpython-312.pyc b/tests/__pycache__/test_kernel_guarantees.cpython-312.pyc new file mode 100644 index 0000000..6f6df56 Binary files /dev/null and b/tests/__pycache__/test_kernel_guarantees.cpython-312.pyc differ diff --git a/tests/__pycache__/test_reference_scenario.cpython-312-pytest-7.4.4.pyc b/tests/__pycache__/test_reference_scenario.cpython-312-pytest-7.4.4.pyc new file mode 100644 index 0000000..dc34970 Binary files /dev/null and b/tests/__pycache__/test_reference_scenario.cpython-312-pytest-7.4.4.pyc differ diff --git a/tests/test_kernel_guarantees.py b/tests/test_kernel_guarantees.py new file mode 100644 index 0000000..44a2547 --- /dev/null +++ b/tests/test_kernel_guarantees.py @@ -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") diff --git a/tests/test_reference_scenario.py b/tests/test_reference_scenario.py new file mode 100644 index 0000000..090e807 --- /dev/null +++ b/tests/test_reference_scenario.py @@ -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() == () diff --git a/workplans/TD-WP-0001-statehub-bootstrap.md b/workplans/TD-WP-0001-statehub-bootstrap.md index 439e319..7c5bc94 100644 --- a/workplans/TD-WP-0001-statehub-bootstrap.md +++ b/workplans/TD-WP-0001-statehub-bootstrap.md @@ -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`. diff --git a/workplans/TD-WP-0002-vertical-spike-crystallization.md b/workplans/TD-WP-0002-vertical-spike-crystallization.md index 8ea00e1..c558ae1 100644 --- a/workplans/TD-WP-0002-vertical-spike-crystallization.md +++ b/workplans/TD-WP-0002-vertical-spike-crystallization.md @@ -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