Skip to content

Python API Reference

Top-level exports

import groundcrew

groundcrew

groundcrew - Deterministic state oracle and semantic action codec for computer-use agents.

ChainVerification(is_valid, chain_length, broken_at=None, errors=list(), summary='') dataclass

Result of verifying a receipt chain.

Attributes:

Name Type Description
is_valid bool

True if the chain is unbroken.
chain_length int

Number of receipts in the chain.
broken_at int | None

Index of the first broken link, or None if the chain is valid.
errors list[str]

List of human-readable error descriptions.
summary str

One-line summary of the verification result.

ClosedLoopError

Bases: ValueError

Raised when the gate refuses empty, unusable, or empty-effect receipts.

GateOutcome(ok, verdict, reason, exit_code, receipt_count=0, total_changed_paths=0, empty_effect_ids=(), dead_path_ids=(), dead_paths=(), human_required=False, risk=None, inventory_count=0, action=None) dataclass

Result of a closed-loop read of groundcrew receipts or destructive gates.

Attributes:

Name Type Description
ok bool

True only when a pipeline may continue (PASS).
verdict str

PASS, FAIL, or FAIL_LOUD.
reason str

Human-readable explanation (always non-empty).
exit_code int

0 PASS, 1 FAIL (action failed), 2 FAIL_LOUD (empty/no side effects).
receipt_count int

Number of receipts examined.
total_changed_paths int

Distinct changed paths across examined receipts.
empty_effect_ids tuple[str, ...]

Receipt IDs that claimed success with zero changes.
dead_path_ids tuple[str, ...]

Receipt IDs with success but paths that fail disk verify.
dead_paths tuple[str, ...]

Sample of claimed paths that are dead on disk.
human_required bool

True when a human must approve before proceeding.
risk str | None

safe, high_risk, or None when not a destructive gate.
inventory_count int

Count of named targets that will be destroyed (if gated).
action str | None

Canonical action / verb that was gated (destructive path).

to_dict()

Serialise for JSON reports (eagle-eyes dogfood, CI artifacts).

Source code in src/groundcrew/closed_loop.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def to_dict(self) -> dict[str, Any]:
    """Serialise for JSON reports (eagle-eyes dogfood, CI artifacts)."""
    return {
        "ok": self.ok,
        "verdict": self.verdict,
        "reason": self.reason,
        "exit_code": self.exit_code,
        "receipt_count": self.receipt_count,
        "total_changed_paths": self.total_changed_paths,
        "empty_effect_ids": list(self.empty_effect_ids),
        "dead_path_ids": list(self.dead_path_ids),
        "dead_paths": list(self.dead_paths),
        "human_required": self.human_required,
        "risk": self.risk,
        "inventory_count": self.inventory_count,
        "action": self.action,
    }

ActionReceipt(spec, before_id, after_id, diff, success, timestamp) dataclass

A verifiable record pairing an action spec with the state change it produced.

ActionSpec(verb, target, params) dataclass

A semantic description of an action: a verb applied to a target with params.

ContentDiff(file_diffs=list(), total_added=0, total_removed=0) dataclass

Aggregated line-level diff across all changed files.

Attributes:

Name Type Description
file_diffs list[FileDiff]

Per-file diff results.
total_added int

Sum of added lines across all files.
total_removed int

Sum of removed lines across all files.

FileDiff(path, before_lines, after_lines, added_lines, removed_lines, unified_diff, is_binary=False, is_approximate=False) dataclass

Line-level diff for a single file.

Attributes:

Name Type Description
path str

Relative path to the file.
before_lines int | None

Number of lines in the before version (0 for new files, None for modified files where before content is unavailable - the before snapshot only stores hashes, not content).
after_lines int

Number of lines in the after version (0 for deleted files).
added_lines int

Number of lines added.
removed_lines int

Number of lines removed.
unified_diff str

Standard unified diff string.
is_binary bool

True if the file was detected as binary.
is_approximate bool

True when the before content is unavailable (modified files). In this case the diff shows all current lines as added because the before state cannot be reconstructed from hashes alone.

Oracle(root, spec=None)

Context manager that snapshots a root before and after a block of work.

Source code in src/groundcrew/oracle.py 
19
20
21
22
23
24
def __init__(self, root: str | Path, spec: ActionSpec | None = None) -> None:
    self.root = Path(root)
    self.spec = spec
    self._before: StateSnapshot | None = None
    self._after: StateSnapshot | None = None
    self._success = True

record(spec)

Build an ActionReceipt for spec from the captured before/after state.

Source code in src/groundcrew/oracle.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
def record(self, spec: ActionSpec) -> ActionReceipt:
    """Build an ActionReceipt for ``spec`` from the captured before/after state."""
    if self._after is None:
        self._after = StateSnapshot.capture(self.root)
    diff = diff_snapshots(self._before, self._after)
    return ActionReceipt(
        spec=spec,
        before_id=self._before.id if self._before else "",
        after_id=self._after.id,
        diff=diff,
        success=self._success,
        timestamp=time.time(),
    )

ReceiptStore(path)

A SQLite-backed store for persisting and retrieving action receipts.

Source code in src/groundcrew/oracle.py 
61
62
63
64
65
66
def __init__(self, path: str | Path) -> None:
    self._path = Path(path)
    self._path.parent.mkdir(parents=True, exist_ok=True)
    self._conn = sqlite3.connect(str(self._path))
    self._conn.execute("CREATE TABLE IF NOT EXISTS receipts (id TEXT PRIMARY KEY, data TEXT)")
    self._conn.commit()

FileState(path, size, sha256) dataclass

The recorded state of a single file: relative path, size, and digest.

SnapshotDiff(snapshot_a_id, snapshot_b_id, added, removed, modified) dataclass

The structural delta between two snapshots: added, removed, modified files.

Attributes:

Name Type Description
added list[FileState]

list[FileState] - files present in b but not in a. Each element is a :class:FileState; use .path to get the relative path string. Example::

      for f in diff.added:
          print(f.path)   # e.g. "subdir/new_file.txt"
removed list[FileState]

list[FileState] - files present in a but not in b. Same type as added; iterate with .path.
modified list[tuple[FileState, FileState]]

list[tuple[FileState, FileState]] - files whose content changed. Each element is (before, after)::

      for before, after in diff.modified:
          print(before.path, before.sha256, "->", after.sha256)

StateSnapshot(id, timestamp, root, files) dataclass

A content-addressed snapshot of every file beneath a root directory.

PlannedToolCall(call_id, name, arguments=dict()) dataclass

One proposed tool invocation (pre-execution).

ToolMisuseReport(validity_ids, over_call_ids, missing_tools, call_count, classes) dataclass

Classified misuse findings for a plan.

ToolSchema(name, required_args=(), arg_types=dict()) dataclass

Required argument contract for a named tool.

DirectoryWatcher(root, authorized_paths=None, interval_seconds=5.0)

Polls a directory for changes and fires callbacks on unexpected mutations.

Typical usage::

watcher = DirectoryWatcher(root="/path/to/dir", interval_seconds=5.0)
watcher.take_baseline()
changes = watcher.check()
if changes:
    print("Unexpected changes:", changes)

Attributes:

Name Type Description
root

The directory being watched.
authorized_paths set[str]

If provided, changes to these paths are considered authorized and will not be reported.
interval_seconds

Polling interval used by :meth:watch.
Source code in src/groundcrew/watcher.py 
30
31
32
33
34
35
36
37
38
39
def __init__(
    self,
    root: Path,
    authorized_paths: list[str] | None = None,
    interval_seconds: float = 5.0,
) -> None:
    self.root = Path(root)
    self.authorized_paths: set[str] = set(authorized_paths or [])
    self.interval_seconds = interval_seconds
    self._baseline: StateSnapshot | None = None

take_baseline()

Capture the current state of the directory as the authorized baseline.

Returns:

Type Description
StateSnapshot

The captured :class:~groundcrew.snapshot.StateSnapshot.
Source code in src/groundcrew/watcher.py 
41
42
43
44
45
46
47
48
def take_baseline(self) -> StateSnapshot:
    """Capture the current state of the directory as the authorized baseline.

    Returns:
        The captured :class:`~groundcrew.snapshot.StateSnapshot`.
    """
    self._baseline = StateSnapshot.capture(self.root)
    return self._baseline

check()

Check for changes since the baseline was taken.

Compares the current directory state against the stored baseline and returns a list of human-readable change descriptions for all changes that are not in :attr:authorized_paths.

Returns:

Type Description
list[str]

List of change descriptions. Empty if no unauthorized changes.

Raises:

Type Description
RuntimeError

If :meth:take_baseline has not been called yet.
Source code in src/groundcrew/watcher.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def check(self) -> list[str]:
    """Check for changes since the baseline was taken.

    Compares the current directory state against the stored baseline and
    returns a list of human-readable change descriptions for all changes
    that are *not* in :attr:`authorized_paths`.

    Returns:
        List of change descriptions. Empty if no unauthorized changes.

    Raises:
        RuntimeError: If :meth:`take_baseline` has not been called yet.
    """
    if self._baseline is None:
        raise RuntimeError("Call take_baseline() before check().")

    current = StateSnapshot.capture(self.root)
    diff = diff_snapshots(self._baseline, current)

    changes: list[str] = []

    for f in diff.added:
        if f.path not in self.authorized_paths:
            changes.append(f"ADDED    {f.path} ({f.size} bytes)")

    for f in diff.removed:
        if f.path not in self.authorized_paths:
            changes.append(f"REMOVED  {f.path}")

    for before, after in diff.modified:
        if before.path not in self.authorized_paths:
            changes.append(f"MODIFIED {before.path} ({before.size} → {after.size} bytes)")

    return changes

watch(callback, max_checks=10)

Poll for changes and invoke callback on unexpected mutations.

Polls up to max_checks times, sleeping :attr:interval_seconds between each poll. This is intentionally non-infinite so it remains testable and composable. Use a loop around :meth:watch for indefinite monitoring.

Parameters:

Name Type Description Default
callback Callable[[list[str]], None]

Called with a list of change description strings whenever unauthorized changes are detected.

 required
max_checks int

Maximum number of polls before returning.

 10
Source code in src/groundcrew/watcher.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def watch(
    self,
    callback: Callable[[list[str]], None],
    max_checks: int = 10,
) -> None:
    """Poll for changes and invoke *callback* on unexpected mutations.

    Polls up to *max_checks* times, sleeping :attr:`interval_seconds`
    between each poll. This is intentionally non-infinite so it remains
    testable and composable. Use a loop around :meth:`watch` for indefinite
    monitoring.

    Args:
        callback: Called with a list of change description strings whenever
            unauthorized changes are detected.
        max_checks: Maximum number of polls before returning.
    """
    if self._baseline is None:
        self.take_baseline()

    for _ in range(max_checks):
        changes = self.check()
        if changes:
            callback(changes)
        time.sleep(self.interval_seconds)

build_chain_report(receipts)

Build a human-readable chain-of-custody report for a sequence of receipts.

The report lists each receipt's action, state transition, outcome, and timestamp, followed by an overall chain verification result.

Parameters:

Name Type Description Default
receipts list[ActionReceipt]

Ordered list of :class:~groundcrew.codec.ActionReceipt objects.

 required

Returns:

Type Description
str

A formatted multi-line string suitable for printing or logging.
Source code in src/groundcrew/chain.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def build_chain_report(receipts: list[ActionReceipt]) -> str:
    """Build a human-readable chain-of-custody report for a sequence of receipts.

    The report lists each receipt's action, state transition, outcome, and timestamp,
    followed by an overall chain verification result.

    Args:
        receipts: Ordered list of :class:`~groundcrew.codec.ActionReceipt` objects.

    Returns:
        A formatted multi-line string suitable for printing or logging.
    """
    lines: list[str] = [
        "Chain-of-Custody Report",
        "=" * 50,
        f"Receipts: {len(receipts)}",
        "",
    ]

    for i, receipt in enumerate(receipts):
        ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(receipt.timestamp))
        status = "SUCCESS" if receipt.success else "FAILURE"
        lines += [
            f"[{i:03d}] {ts}  {status}",
            f"      Action : {receipt.spec.verb} → {receipt.spec.target}",
            f"      Before : {receipt.before_id}",
            f"      After  : {receipt.after_id}",
            f"      Receipt: {receipt.id}",
            f"      Changed: +{len(receipt.diff.added)} files, "
            f"-{len(receipt.diff.removed)} files, "
            f"~{len(receipt.diff.modified)} files",
            "",
        ]

    verification = verify_chain(receipts)
    lines += [
        "-" * 50,
        f"Verification: {verification.summary}",
    ]

    if not verification.is_valid:
        for err in verification.errors:
            lines.append(f"  ERROR: {err}")

    return "\n".join(lines)

verify_chain(receipts)

Verify that a sequence of receipts forms an unbroken chain.

The chain is valid if for every consecutive pair: receipts[n].after_id == receipts[n+1].before_id

An empty list is considered trivially valid (length 0). A single-receipt list is also valid (nothing to check).

Parameters:

Name Type Description Default
receipts list[ActionReceipt]

Ordered list of :class:~groundcrew.codec.ActionReceipt objects.

 required

Returns:

Name Type Description
A ChainVerification

class:ChainVerification with validity, broken index, and errors.
Source code in src/groundcrew/chain.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def verify_chain(receipts: list[ActionReceipt]) -> ChainVerification:
    """Verify that a sequence of receipts forms an unbroken chain.

    The chain is valid if for every consecutive pair:
    ``receipts[n].after_id == receipts[n+1].before_id``

    An empty list is considered trivially valid (length 0).
    A single-receipt list is also valid (nothing to check).

    Args:
        receipts: Ordered list of :class:`~groundcrew.codec.ActionReceipt` objects.

    Returns:
        A :class:`ChainVerification` with validity, broken index, and errors.
    """
    n = len(receipts)

    if n <= 1:
        return ChainVerification(
            is_valid=True,
            chain_length=n,
            broken_at=None,
            errors=[],
            summary=f"Chain valid: {n} receipt(s), nothing to verify."
            if n == 0
            else "Chain valid: single receipt.",
        )

    errors: list[str] = []
    broken_at: int | None = None

    for i in range(n - 1):
        expected_before = receipts[i].after_id
        actual_before = receipts[i + 1].before_id
        if expected_before != actual_before:
            if broken_at is None:
                broken_at = i + 1
            errors.append(
                f"Link broken at index {i + 1}: "
                f"receipt[{i}].after_id={expected_before!r} "
                f"!= receipt[{i + 1}].before_id={actual_before!r}"
            )

    is_valid = len(errors) == 0
    summary = (
        f"Chain valid: {n} receipt(s) form an unbroken chain."
        if is_valid
        else f"Chain BROKEN at index {broken_at}: {len(errors)} error(s) found."
    )

    return ChainVerification(
        is_valid=is_valid,
        chain_length=n,
        broken_at=broken_at,
        errors=errors,
        summary=summary,
    )

assert_not_destructive(verb='', **kwargs)

Raise :class:ClosedLoopError unless :func:gate_destructive is ok.

Source code in src/groundcrew/closed_loop.py 
656
657
658
659
660
661
662
663
664
def assert_not_destructive(
    verb: str = "",
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_destructive` is ok."""
    outcome = gate_destructive(verb, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

assert_side_effects(source, **kwargs)

Gate receipts and raise :class:ClosedLoopError unless outcome is ok.

Source code in src/groundcrew/closed_loop.py 
397
398
399
400
401
402
403
404
405
def assert_side_effects(
    source: ReceiptStore | Sequence[ActionReceipt] | str | Path,
    **kwargs: Any,
) -> GateOutcome:
    """Gate receipts and raise :class:`ClosedLoopError` unless outcome is ok."""
    outcome = gate_receipts(source, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

dead_paths_for_receipt(receipt, root)

Return claimed side-effect paths that do not match the live workspace.

D-GCROOT / L10 harden: a success receipt can invent FileState rows so changed_paths is non-empty while nothing real happened. When a workspace root is known, every path in the structural diff must match disk:

  • added / modified: file must exist under root
  • removed: file must not exist under root

Returns the list of dead (mismatch) relative paths (may be empty).

Source code in src/groundcrew/closed_loop.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def dead_paths_for_receipt(receipt: ActionReceipt, root: str | Path) -> list[str]:
    """Return claimed side-effect paths that do not match the live workspace.

    D-GCROOT / L10 harden: a success receipt can invent ``FileState`` rows so
    ``changed_paths`` is non-empty while nothing real happened. When a workspace
    ``root`` is known, every path in the structural diff must match disk:

    - **added** / **modified**: file must exist under ``root``
    - **removed**: file must *not* exist under ``root``

    Returns the list of dead (mismatch) relative paths (may be empty).
    """
    base = Path(root)
    dead: list[str] = []
    diff: SnapshotDiff = receipt.diff

    for f in diff.added:
        if not (base / f.path).is_file():
            dead.append(f.path)

    for before, after in diff.modified:
        # Prefer after.path (post-change); fall back to before.path
        path = after.path or before.path
        if not (base / path).is_file():
            dead.append(path)

    for f in diff.removed:
        if (base / f.path).is_file():
            # Claimed removed but still present → dead claim
            dead.append(f.path)

    # Deduplicate while preserving order
    seen: set[str] = set()
    ordered: list[str] = []
    for p in dead:
        if p not in seen:
            seen.add(p)
            ordered.append(p)
    return ordered

gate_destructive(verb='', *, target='', sql=None, command=None, params=None, inventory=None, approved=False, approval_token=None, environment='production', require_inventory=True)

Block unattended destructive tools (Replit DB wipe / Antigravity / AgentWard).

Load-bearing controls (all required for destructive ops in strict envs):

  1. Classify - verb / SQL / shell must be detected as destructive.
  2. Inventory - named targets that will be destroyed (tables, paths, DBs). Empty inventory = agent does not know what it is wiping → FAIL_LOUD.
  3. Approval - human token / approved flag. Missing → FAIL_LOUD (human_required=True).

Non-destructive calls PASS without inventory or approval.

Parameters:

Name Type Description Default
verb str

Tool verb (e.g. db_wipe, drop, execute_sql).

 ''
target str

Logical target name (db, table, path).

 ''
sql str | None

Free-form SQL if the tool accepts queries.

 None
command str | None

Shell command string if applicable.

 None
params dict[str, Any] | None

Extra tool params (may embed sql / command / query).

 None
inventory Sequence[str] | None

Explicit list of objects that will be destroyed.

 None
approved bool

True when a human (or humanproof session) already approved.

 False
approval_token str | None

Opaque owner-issued token id/secret handle.

 None
environment str

production/staging always require approval; dev/test still require inventory for destructive ops.

 'production'
require_inventory bool

If True (default), destructive ops need non-empty inventory even when approved.

 True

Returns:

Type Description
GateOutcome

class:GateOutcome - callers must not execute the tool unless ok.
Source code in src/groundcrew/closed_loop.py 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
def gate_destructive(
    verb: str = "",
    *,
    target: str = "",
    sql: str | None = None,
    command: str | None = None,
    params: dict[str, Any] | None = None,
    inventory: Sequence[str] | None = None,
    approved: bool = False,
    approval_token: str | None = None,
    environment: str = "production",
    require_inventory: bool = True,
) -> GateOutcome:
    """Block unattended destructive tools (Replit DB wipe / Antigravity / AgentWard).

    Load-bearing controls (all required for destructive ops in strict envs):

    1. **Classify** - verb / SQL / shell must be detected as destructive.
    2. **Inventory** - named targets that will be destroyed (tables, paths, DBs).
       Empty inventory = agent does not know what it is wiping → FAIL_LOUD.
    3. **Approval** - human token / approved flag. Missing → FAIL_LOUD
       (``human_required=True``).

    Non-destructive calls PASS without inventory or approval.

    Args:
        verb: Tool verb (e.g. ``db_wipe``, ``drop``, ``execute_sql``).
        target: Logical target name (db, table, path).
        sql: Free-form SQL if the tool accepts queries.
        command: Shell command string if applicable.
        params: Extra tool params (may embed ``sql`` / ``command`` / ``query``).
        inventory: Explicit list of objects that will be destroyed.
        approved: True when a human (or humanproof session) already approved.
        approval_token: Opaque owner-issued token id/secret handle.
        environment: ``production``/``staging`` always require approval;
            ``dev``/``test`` still require inventory for destructive ops.
        require_inventory: If True (default), destructive ops need non-empty
            inventory even when approved.

    Returns:
        :class:`GateOutcome` - callers must not execute the tool unless ``ok``.
    """
    v = _canonical_verb(verb)
    action = v or (f"sql:{sql[:40]}" if sql else (f"cmd:{command[:40]}" if command else ""))
    env = (environment or "production").strip().lower()
    strict = env in _STRICT_ENVIRONMENTS or env == ""

    # Completely empty call - nothing to gate.
    if not v and not (sql and str(sql).strip()) and not (command and str(command).strip()):
        p = params or {}
        if not p.get("sql") and not p.get("query") and not p.get("command") and not p.get("shell"):
            return _fail_loud(
                "DB-WIPE: empty tool call - no verb/sql/command to classify "
                "(cannot gate a phantom destructive action)",
                human_required=True,
                risk="high_risk",
                action=action or None,
            )

    destructive = is_destructive(verb, target=target, sql=sql, command=command, params=params)
    inv = _inventory_list(inventory)
    # Implicit inventory from target only when inventory was omitted (None),
    # not when the caller explicitly passed an empty list (declared no targets).
    if inventory is None and not inv and target and str(target).strip():
        inv = [str(target).strip()]

    if not destructive:
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=f"non-destructive tool call action={action!r} env={env}",
            exit_code=0,
            human_required=False,
            risk="safe",
            inventory_count=len(inv),
            action=action or None,
        )

    # --- Destructive path ---
    if require_inventory and len(inv) == 0:
        return _fail_loud(
            "DB-WIPE: destructive action without inventory - "
            f"action={action!r} env={env} "
            "(Replit/AgentWard class: agent must name tables/paths/DBs before wipe)",
            human_required=True,
            risk="high_risk",
            inventory_count=0,
            action=action or None,
        )

    needs_approval = strict or env not in {"dev", "development", "test", "local", "ci"}
    has_auth = _has_approval(approved=approved, approval_token=approval_token)
    if needs_approval and not has_auth:
        return _fail_loud(
            "DB-WIPE: destructive action without human approval - "
            f"action={action!r} env={env} inventory={inv[:10]} "
            "(public: Replit AI production DB deletion; Antigravity wipe; "
            "AgentWard post-incident). Call humanproof.gate_approval first.",
            human_required=True,
            risk="high_risk",
            inventory_count=len(inv),
            action=action or None,
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"destructive action authorised: action={action!r} env={env} "
            f"inventory_count={len(inv)} approved={has_auth}"
        ),
        exit_code=0,
        human_required=False,
        risk="high_risk",
        inventory_count=len(inv),
        action=action or None,
    )

gate_destructive_receipt(receipt, *, inventory=None, approved=False, approval_token=None, environment='production', require_inventory=True)

Gate an :class:ActionReceipt for destructive verbs (pre- or post-exec).

Uses receipt.spec.verb/target/params plus optional explicit inventory. When inventory is omitted, falls back to receipt.diff.changed_paths (filesystem-class wipes) then spec.target.

Source code in src/groundcrew/closed_loop.py 
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
def gate_destructive_receipt(
    receipt: ActionReceipt,
    *,
    inventory: Sequence[str] | None = None,
    approved: bool = False,
    approval_token: str | None = None,
    environment: str = "production",
    require_inventory: bool = True,
) -> GateOutcome:
    """Gate an :class:`ActionReceipt` for destructive verbs (pre- or post-exec).

    Uses ``receipt.spec.verb/target/params`` plus optional explicit inventory.
    When inventory is omitted, falls back to ``receipt.diff.changed_paths``
    (filesystem-class wipes) then ``spec.target``.
    """
    if not isinstance(receipt, ActionReceipt):
        return _fail_loud(
            "DB-WIPE: gate_destructive_receipt requires ActionReceipt",
            human_required=True,
            risk="high_risk",
        )
    spec: ActionSpec = receipt.spec
    inv = inventory
    if inv is None:
        paths = list(receipt.diff.changed_paths)
        inv = paths if paths else None
    return gate_destructive(
        spec.verb,
        target=spec.target,
        params=spec.params,
        sql=spec.params.get("sql") if isinstance(spec.params, dict) else None,
        command=spec.params.get("command") if isinstance(spec.params, dict) else None,
        inventory=inv,
        approved=approved,
        approval_token=approval_token,
        environment=environment,
        require_inventory=require_inventory,
    )

gate_receipts(source, *, require_side_effects=True, require_any_success=True, root=None, verify_disk=None)

Read receipts and fail loudly when success has no filesystem side effects.

Parameters:

Name Type Description Default
source ReceiptStore | Sequence[ActionReceipt] | str | Path

Open :class:ReceiptStore, path to a receipts SQLite db, or an in-memory sequence of :class:ActionReceipt.

 required
require_side_effects bool

If True (L10 default), any receipt with success=True and zero changed_paths is FAIL_LOUD.

 True
require_any_success bool

If True, a non-empty set of receipts where every receipt has success=False is FAIL (not FAIL_LOUD - evidence of attempted work that failed).

 True
root str | Path | None

Workspace directory for D-GCROOT dead-path verification. When set (or when verify_disk is True with a root), success receipts whose claimed paths do not match the live tree are FAIL_LOUD.

 None
verify_disk bool | None

Force on/off disk verification. Default: True when root is provided, False otherwise.

 None

Returns:

Type Description
GateOutcome

class:GateOutcome - callers should sys.exit(outcome.exit_code).
Source code in src/groundcrew/closed_loop.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
def gate_receipts(
    source: ReceiptStore | Sequence[ActionReceipt] | str | Path,
    *,
    require_side_effects: bool = True,
    require_any_success: bool = True,
    root: str | Path | None = None,
    verify_disk: bool | None = None,
) -> GateOutcome:
    """Read receipts and fail loudly when success has no filesystem side effects.

    Args:
        source: Open :class:`ReceiptStore`, path to a receipts SQLite db, or an
            in-memory sequence of :class:`ActionReceipt`.
        require_side_effects: If True (L10 default), any receipt with
            ``success=True`` and zero ``changed_paths`` is FAIL_LOUD.
        require_any_success: If True, a non-empty set of receipts where every
            receipt has ``success=False`` is FAIL (not FAIL_LOUD - evidence of
            attempted work that failed).
        root: Workspace directory for D-GCROOT dead-path verification. When set
            (or when ``verify_disk`` is True with a root), success receipts whose
            claimed paths do not match the live tree are FAIL_LOUD.
        verify_disk: Force on/off disk verification. Default: True when ``root``
            is provided, False otherwise.

    Returns:
        :class:`GateOutcome` - callers should ``sys.exit(outcome.exit_code)``.
    """
    do_disk = verify_disk if verify_disk is not None else (root is not None)
    if do_disk and root is None:
        return _fail_loud(
            "D-GCROOT: verify_disk=True requires root= workspace path "
            "(cannot prove side effects without a tree)"
        )

    owns = False
    store: ReceiptStore | None = None
    try:
        if isinstance(source, ReceiptStore):
            receipts = list(source.list_receipts())
        elif isinstance(source, (str, Path)):
            path = Path(source)
            if not path.is_file():
                return _fail_loud(f"receipt store not found: {path}")
            try:
                store = ReceiptStore(path)
                owns = True
                receipts = list(store.list_receipts())
            except Exception as exc:
                return _fail_loud(f"open receipt store failed: {exc.__class__.__name__}: {exc}")
        else:
            receipts = list(source)

        if len(receipts) == 0:
            return _fail_loud(
                "empty receipts - no load-bearing filesystem side effects to gate "
                "(write-only ornament / L10)"
            )

        empty_effect: list[str] = []
        dead_effect: list[str] = []
        dead_path_samples: list[str] = []
        failed: list[str] = []
        all_paths: set[str] = set()
        success_count = 0

        for r in receipts:
            paths = set(r.diff.changed_paths)
            all_paths |= paths
            if r.success:
                success_count += 1
                if require_side_effects and len(paths) == 0:
                    empty_effect.append(r.id)
                elif do_disk and root is not None and require_side_effects:
                    dead = dead_paths_for_receipt(r, root)
                    if dead:
                        dead_effect.append(r.id)
                        for p in dead:
                            if p not in dead_path_samples:
                                dead_path_samples.append(p)
            else:
                failed.append(r.id)

        total_changed = len(all_paths)
        empty_ids = tuple(empty_effect)
        dead_ids = tuple(dead_effect)
        dead_paths_t = tuple(dead_path_samples[:20])

        if empty_effect:
            return _fail_loud(
                "L10: success with empty side effects - "
                f"receipt_ids={list(empty_effect)} "
                "(success requires non-empty changed_paths)",
                receipt_count=len(receipts),
                total_changed_paths=total_changed,
                empty_effect_ids=empty_ids,
            )

        if dead_effect:
            return _fail_loud(
                "D-GCROOT: success with dead/phantom paths - "
                f"receipt_ids={list(dead_effect)} "
                f"dead_paths={dead_path_samples[:10]} "
                "(claimed side effects do not match workspace root)",
                receipt_count=len(receipts),
                total_changed_paths=total_changed,
                empty_effect_ids=(),
                dead_path_ids=dead_ids,
                dead_paths=dead_paths_t,
            )

        if require_any_success and success_count == 0:
            return GateOutcome(
                ok=False,
                verdict="FAIL",
                reason=(f"all receipts failed action (count={len(receipts)} failed_ids={failed})"),
                exit_code=1,
                receipt_count=len(receipts),
                total_changed_paths=total_changed,
                empty_effect_ids=(),
            )

        if require_side_effects and total_changed == 0:
            return _fail_loud(
                "L10: no filesystem side effects across receipts "
                f"(count={len(receipts)} changed_paths=0)",
                receipt_count=len(receipts),
                total_changed_paths=0,
            )

        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason=(
                f"receipts ok: count={len(receipts)} success={success_count} "
                f"changed_paths={total_changed}" + (f" disk_verified={root}" if do_disk else "")
            ),
            exit_code=0,
            receipt_count=len(receipts),
            total_changed_paths=total_changed,
            empty_effect_ids=(),
        )
    finally:
        if owns and store is not None:
            with contextlib.suppress(Exception):
                store.close()

is_destructive(verb='', *, target='', sql=None, command=None, params=None)

Classify a tool call as destructive (irreversible data/schema/file loss).

Checks (any match → destructive): 1. Verb in :data:DESTRUCTIVE_VERBS (exact or prefix verb:scope) 2. SQL payload via :func:sql_is_destructive 3. Shell command via :func:shell_is_destructive 4. params['sql'] / params['command'] / params['query']

Source code in src/groundcrew/closed_loop.py 
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
def is_destructive(
    verb: str = "",
    *,
    target: str = "",
    sql: str | None = None,
    command: str | None = None,
    params: dict[str, Any] | None = None,
) -> bool:
    """Classify a tool call as destructive (irreversible data/schema/file loss).

    Checks (any match → destructive):
      1. Verb in :data:`DESTRUCTIVE_VERBS` (exact or prefix ``verb:scope``)
      2. SQL payload via :func:`sql_is_destructive`
      3. Shell command via :func:`shell_is_destructive`
      4. ``params['sql']`` / ``params['command']`` / ``params['query']``
    """
    v = _canonical_verb(verb)
    if v:
        # Allow "drop:table", "db_wipe:prod", "delete:users"
        base = v.split(":", 1)[0]
        if base in DESTRUCTIVE_VERBS or v in DESTRUCTIVE_VERBS:
            return True

    p = params or {}
    sql_blob = sql if sql is not None else p.get("sql") or p.get("query")
    if sql_blob and sql_is_destructive(str(sql_blob)):
        return True

    cmd_blob = command if command is not None else p.get("command") or p.get("shell")
    if cmd_blob and shell_is_destructive(str(cmd_blob)):
        return True

    # Target alone is not enough (e.g. verb=read target=users) - only when
    # combined with destructive verb, already handled above.
    _ = target
    return False

shell_is_destructive(command)

Return True if command looks like rm -rf / shred / dd overwrite class.

Source code in src/groundcrew/closed_loop.py 
425
426
427
428
429
def shell_is_destructive(command: str) -> bool:
    """Return True if *command* looks like rm -rf / shred / dd overwrite class."""
    if not command or not str(command).strip():
        return False
    return _SHELL_DESTRUCTIVE_RE.search(str(command)) is not None

sql_is_destructive(sql)

Return True if sql contains irreversible DROP/TRUNCATE/DELETE-class ops.

Public incidents (Replit AI production DB wipe, AgentWard file wipe) start with free-form SQL tools that accept any string. Classifiers must refuse before execution - not after a success receipt is written.

Source code in src/groundcrew/closed_loop.py 
413
414
415
416
417
418
419
420
421
422
def sql_is_destructive(sql: str) -> bool:
    """Return True if *sql* contains irreversible DROP/TRUNCATE/DELETE-class ops.

    Public incidents (Replit AI production DB wipe, AgentWard file wipe) start
    with free-form SQL tools that accept any string. Classifiers must refuse
    before execution - not after a success receipt is written.
    """
    if not sql or not str(sql).strip():
        return False
    return _SQL_DESTRUCTIVE_RE.search(str(sql)) is not None

analyze_tool_misuse(calls, *, schemas=None, tools_required=False, required_tools=None, tools_forbidden=False, max_calls=None)

Classify validity / over-calling / missing failures on a tool plan.

Parameters:

Name Type Description Default
calls Sequence[PlannedToolCall | dict[str, Any]] | None

Planned tool calls (may be empty).

 required
schemas Sequence[ToolSchema | dict[str, Any]] | Mapping[str, ToolSchema | dict[str, Any]] | None

Per-tool required-arg contracts (list or name→schema map).

 None
tools_required bool

If True and calls empty → missing class.

 False
required_tools Iterable[str] | None

Tool names that must appear at least once.

 None
tools_forbidden bool

If True, any call is over-calling (answer-only turn).

 False
max_calls int | None

Soft cap; excess calls tagged over-calling.

 None
Source code in src/groundcrew/tool_misuse.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def analyze_tool_misuse(
    calls: Sequence[PlannedToolCall | dict[str, Any]] | None,
    *,
    schemas: (
        Sequence[ToolSchema | dict[str, Any]] | Mapping[str, ToolSchema | dict[str, Any]] | None
    ) = None,
    tools_required: bool = False,
    required_tools: Iterable[str] | None = None,
    tools_forbidden: bool = False,
    max_calls: int | None = None,
) -> ToolMisuseReport:
    """Classify validity / over-calling / missing failures on a tool plan.

    Args:
        calls: Planned tool calls (may be empty).
        schemas: Per-tool required-arg contracts (list or name→schema map).
        tools_required: If True and calls empty → missing class.
        required_tools: Tool names that must appear at least once.
        tools_forbidden: If True, any call is over-calling (answer-only turn).
        max_calls: Soft cap; excess calls tagged over-calling.
    """
    planned: list[PlannedToolCall] = []
    if calls:
        for i, c in enumerate(calls):
            planned.append(_as_call(c, i))

    schema_map: dict[str, ToolSchema] = {}
    if schemas is not None:
        if isinstance(schemas, Mapping):
            for k, v in schemas.items():
                if isinstance(v, ToolSchema):
                    s = v
                elif isinstance(v, dict):
                    d = dict(v)
                    if not d.get("name"):
                        d["name"] = str(k)
                    s = _as_schema(d)
                else:
                    s = ToolSchema(name=str(k))
                schema_map[s.name] = s
                schema_map[str(k)] = s
        else:
            for item in schemas:
                sc = _as_schema(item)
                schema_map[sc.name] = sc

    validity: list[str] = []
    for c in planned:
        sch = schema_map.get(c.name) or schema_map.get(c.name.lower())
        if not call_is_valid(c, sch):
            validity.append(c.call_id)

    over: list[str] = []
    if tools_forbidden and planned:
        over.extend(c.call_id for c in planned)
    if max_calls is not None and max_calls >= 0 and len(planned) > max_calls:
        over.extend(c.call_id for c in planned[max_calls:])

    missing: list[str] = []
    if tools_required and not planned:
        missing.append("*")
    if required_tools:
        present = {c.name for c in planned}
        for t in required_tools:
            name = str(t).strip()
            if name and name not in present:
                missing.append(name)

    classes: list[MisuseClass] = []
    if validity:
        classes.append("validity")
    if over:
        classes.append("over_calling")
    if missing:
        classes.append("missing")
    if not classes:
        classes.append("ok")

    return ToolMisuseReport(
        validity_ids=tuple(dict.fromkeys(validity)),
        over_call_ids=tuple(dict.fromkeys(over)),
        missing_tools=tuple(dict.fromkeys(missing)),
        call_count=len(planned),
        classes=tuple(classes),
    )

assert_tool_misuse_ok(calls=None, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_tool_misuse is ok.

Source code in src/groundcrew/tool_misuse.py 
355
356
357
358
359
360
361
362
363
def assert_tool_misuse_ok(
    calls: Sequence[PlannedToolCall | dict[str, Any]] | None = None,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_tool_misuse` is ok."""
    outcome = gate_tool_misuse(calls, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

call_is_valid(call, schema)

True when required args present and types match schema (if provided).

Source code in src/groundcrew/tool_misuse.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def call_is_valid(call: PlannedToolCall, schema: ToolSchema | None) -> bool:
    """True when required args present and types match schema (if provided)."""
    if schema is None:
        # no schema - only reject completely empty name
        return bool(call.name)
    args = call.arguments or {}
    for key in schema.required_args:
        if key not in args or args[key] is None or args[key] == "":
            return False
        expected = schema.arg_types.get(key)
        if expected and not _type_ok(args[key], expected):
            return False
    for key, expected in schema.arg_types.items():
        if (
            key in args
            and args[key] is not None
            and args[key] != ""
            and not _type_ok(args[key], expected)
        ):
            return False
    return True

gate_tool_misuse(calls=None, *, schemas=None, tools_required=False, required_tools=None, tools_forbidden=False, max_calls=None, refuse_validity=True, refuse_over_calling=True, refuse_missing=True)

Refuse plans with PRISMS-class tool misuse (arXiv 2608.00218).

Rules:

  • Invalid args (validity) → FAIL
  • Over-calling when tools forbidden / over max_calls → FAIL
  • Missing required tools / empty when tools_required → FAIL_LOUD (missing is pre-generation boundary class - empty inventory)
  • Clean plan → PASS
Source code in src/groundcrew/tool_misuse.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def gate_tool_misuse(
    calls: Sequence[PlannedToolCall | dict[str, Any]] | None = None,
    *,
    schemas: (
        Sequence[ToolSchema | dict[str, Any]] | Mapping[str, ToolSchema | dict[str, Any]] | None
    ) = None,
    tools_required: bool = False,
    required_tools: Iterable[str] | None = None,
    tools_forbidden: bool = False,
    max_calls: int | None = None,
    refuse_validity: bool = True,
    refuse_over_calling: bool = True,
    refuse_missing: bool = True,
) -> GateOutcome:
    """Refuse plans with PRISMS-class tool misuse (arXiv 2608.00218).

    Rules:

    * Invalid args (validity) → **FAIL**
    * Over-calling when tools forbidden / over max_calls → **FAIL**
    * Missing required tools / empty when tools_required → **FAIL_LOUD**
      (missing is pre-generation boundary class - empty inventory)
    * Clean plan → **PASS**
    """
    try:
        report = analyze_tool_misuse(
            calls,
            schemas=schemas,
            tools_required=tools_required,
            required_tools=required_tools,
            tools_forbidden=tools_forbidden,
            max_calls=max_calls,
        )
    except (TypeError, ValueError) as exc:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=f"TOOL-MISUSE: invalid plan payload: {exc}",
            exit_code=2,
            human_required=True,
        )

    if refuse_missing and report.missing_tools:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=(
                f"TOOL-MISUSE/missing: required tool call(s) omitted "
                f"missing={list(report.missing_tools)[:8]} call_count={report.call_count} "
                f"- refuse answer-only path when tools are needed "
                f"(arXiv 2608.00218 PRISMS missing class)"
            ),
            exit_code=2,
            human_required=True,
            receipt_count=report.call_count,
            action="missing",
            risk="high_risk",
        )

    if refuse_validity and report.validity_ids:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"TOOL-MISUSE/validity: {len(report.validity_ids)} call(s) with "
                f"invalid/incomplete arguments ids={list(report.validity_ids)[:8]} "
                f"- refuse execution (PRISMS validity class)"
            ),
            exit_code=1,
            human_required=True,
            receipt_count=report.call_count,
            action="validity",
            risk="high_risk",
            empty_effect_ids=report.validity_ids[:20],
        )

    if refuse_over_calling and report.over_call_ids:
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"TOOL-MISUSE/over_calling: {len(report.over_call_ids)} unnecessary "
                f"call(s) ids={list(report.over_call_ids)[:8]} "
                f"(tools_forbidden={tools_forbidden} max_calls={max_calls}) - "
                f"refuse surplus tool use (PRISMS over-calling class)"
            ),
            exit_code=1,
            human_required=False,
            receipt_count=report.call_count,
            action="over_calling",
            risk="safe",
            empty_effect_ids=report.over_call_ids[:20],
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(f"TOOL-MISUSE ok: calls={report.call_count} classes={list(report.classes)}"),
        exit_code=0,
        human_required=False,
        receipt_count=report.call_count,
        action="ok",
        risk="safe",
    )