diff --git a/skills/openclaw-native/expansion-grant-guard/SKILL.md b/skills/openclaw-native/expansion-grant-guard/SKILL.md new file mode 100644 index 0000000..237064f --- /dev/null +++ b/skills/openclaw-native/expansion-grant-guard/SKILL.md @@ -0,0 +1,142 @@ +--- +name: expansion-grant-guard +version: "1.0" +category: openclaw-native +description: YAML-based delegation grant ledger — issues, validates, and tracks scoped permission grants for sub-agent expansions with token budgets and auto-expiry. +stateful: true +--- + +# Expansion Grant Guard + +## What it does + +When an agent delegates work to a sub-agent (or expands context via DAG recall), it needs a controlled way to grant scoped permissions. Expansion Grant Guard maintains a YAML-based grant ledger that issues time-limited, token-budgeted grants — ensuring sub-agent operations stay within defined boundaries. + +Inspired by [lossless-claw](https://github.com/Martian-Engineering/lossless-claw)'s delegation grant system, where a parent agent issues a signed grant specifying what a sub-agent can access, how many tokens it may consume, and when the grant expires. + +## When to invoke + +- Before any sub-agent expansion or delegation — issue a grant first +- When a sub-agent requests resources — validate the grant before proceeding +- When checking token budgets — verify remaining budget in the grant +- Periodically to clean up expired grants — auto-expiry sweep + +## How to use + +```bash +python3 guard.py --issue --scope "dag-recall" --budget 4000 --ttl 30 # Issue a grant +python3 guard.py --validate # Check if grant is valid +python3 guard.py --consume --tokens 500 # Record token usage +python3 guard.py --revoke # Revoke a grant early +python3 guard.py --list # List all active grants +python3 guard.py --sweep # Clean up expired grants +python3 guard.py --audit # Full audit log +python3 guard.py --stats # Grant statistics +python3 guard.py --status # Current status summary +python3 guard.py --format json # Machine-readable output +``` + +## Grant structure + +Each grant is a YAML entry in the ledger: + +```yaml +grant_id: "g-20260316-001" +scope: "dag-recall" # What the grant allows +issued_at: "2026-03-16T10:00:00Z" +expires_at: "2026-03-16T10:30:00Z" +token_budget: 4000 # Max tokens allowed +tokens_consumed: 1250 # Tokens used so far +status: active # active | expired | revoked | exhausted +issuer: "parent-session" +metadata: + query: "auth migration" + reason: "Recalling auth decisions for new implementation" +``` + +## Grant lifecycle + +``` +Issue → Active → { Consumed | Expired | Revoked | Exhausted } + │ │ + │ ├─ tokens_consumed < budget → still active + │ ├─ tokens_consumed >= budget → exhausted + │ ├─ now > expires_at → expired + │ └─ explicit revoke → revoked + │ + └─ Validation checks: status=active AND not expired AND budget remaining +``` + +## Procedure + +**Step 1 — Issue a grant before expansion** + +```bash +python3 guard.py --issue --scope "dag-recall" --budget 4000 --ttl 30 +``` + +Output: +``` +Grant Issued +───────────────────────────────────────────── + Grant ID: g-20260316-001 + Scope: dag-recall + Token budget: 4,000 + Expires: 2026-03-16T10:30:00Z (in 30 min) + Status: active +``` + +**Step 2 — Validate before consuming resources** + +```bash +python3 guard.py --validate g-20260316-001 +``` + +Returns status, remaining budget, and time until expiry. Non-zero exit code if invalid. + +**Step 3 — Record token consumption** + +```bash +python3 guard.py --consume g-20260316-001 --tokens 1250 +``` + +Deducts from the grant's remaining budget. Fails if exceeding budget. + +**Step 4 — Sweep expired grants** + +```bash +python3 guard.py --sweep +``` + +Marks all expired grants and cleans up the active list. + +## Scope types + +- `dag-recall` — Permission to walk DAG nodes and assemble answers +- `session-search` — Permission to search session persistence database +- `file-access` — Permission to read externalized large files +- `context-expand` — Permission to expand context window with external data +- `tool-invoke` — Permission to invoke external tools/MCP servers +- Custom scopes accepted — any string is valid + +## Integration with other skills + +- **dag-recall**: Should issue a grant before expanding DAG nodes; checks budget during expansion +- **session-persistence**: Grant-gated search — validate grant before querying message database +- **large-file-interceptor**: Grant-gated file restore — validate before loading large files back +- **context-assembly-scorer**: Token budget tracking feeds into assembly scoring + +## State + +Grant ledger and audit log stored in `~/.openclaw/skill-state/expansion-grant-guard/state.yaml`. + +Fields: `active_grants`, `total_issued`, `total_expired`, `total_revoked`, `total_exhausted`, `total_tokens_granted`, `total_tokens_consumed`, `grant_history`. + +## Notes + +- Uses Python's built-in modules only — no external dependencies +- Grant IDs are timestamped and sequential within a day +- Ledger file is append-friendly YAML — safe for concurrent reads +- Expired grants kept in history for audit; active list stays clean after sweep +- Default TTL is 30 minutes; max TTL is 24 hours +- Token budget is advisory — enforcement depends on consuming skill cooperation diff --git a/skills/openclaw-native/expansion-grant-guard/STATE_SCHEMA.yaml b/skills/openclaw-native/expansion-grant-guard/STATE_SCHEMA.yaml new file mode 100644 index 0000000..f1d586e --- /dev/null +++ b/skills/openclaw-native/expansion-grant-guard/STATE_SCHEMA.yaml @@ -0,0 +1,38 @@ +version: "1.0" +description: Grant ledger, consumption tracking, and audit history. +fields: + active_grants: + type: list + description: Currently active grants + items: + grant_id: { type: string } + scope: { type: string } + issued_at: { type: datetime } + expires_at: { type: datetime } + token_budget: { type: integer } + tokens_consumed: { type: integer } + status: { type: enum, values: [active, expired, revoked, exhausted] } + issuer: { type: string } + total_issued: + type: integer + total_expired: + type: integer + total_revoked: + type: integer + total_exhausted: + type: integer + total_tokens_granted: + type: integer + total_tokens_consumed: + type: integer + grant_history: + type: list + description: Rolling log of completed grants (last 50) + items: + grant_id: { type: string } + scope: { type: string } + issued_at: { type: datetime } + closed_at: { type: datetime } + token_budget: { type: integer } + tokens_consumed: { type: integer } + final_status: { type: string } diff --git a/skills/openclaw-native/expansion-grant-guard/example-state.yaml b/skills/openclaw-native/expansion-grant-guard/example-state.yaml new file mode 100644 index 0000000..df92bff --- /dev/null +++ b/skills/openclaw-native/expansion-grant-guard/example-state.yaml @@ -0,0 +1,75 @@ +# Example runtime state for expansion-grant-guard +active_grants: + - grant_id: "g-20260316-003" + scope: "dag-recall" + issued_at: "2026-03-16T10:30:00Z" + expires_at: "2026-03-16T11:00:00Z" + token_budget: 4000 + tokens_consumed: 1250 + status: active + issuer: "parent-session" + - grant_id: "g-20260316-004" + scope: "session-search" + issued_at: "2026-03-16T10:45:00Z" + expires_at: "2026-03-16T11:15:00Z" + token_budget: 2000 + tokens_consumed: 0 + status: active + issuer: "parent-session" +total_issued: 12 +total_expired: 6 +total_revoked: 1 +total_exhausted: 3 +total_tokens_granted: 48000 +total_tokens_consumed: 28750 +grant_history: + - grant_id: "g-20260316-001" + scope: "dag-recall" + issued_at: "2026-03-16T09:00:00Z" + closed_at: "2026-03-16T09:22:15Z" + token_budget: 4000 + tokens_consumed: 3840 + final_status: exhausted + - grant_id: "g-20260316-002" + scope: "context-expand" + issued_at: "2026-03-16T09:30:00Z" + closed_at: "2026-03-16T10:00:00Z" + token_budget: 8000 + tokens_consumed: 2100 + final_status: expired +# ── Walkthrough ────────────────────────────────────────────────────────────── +# python3 guard.py --issue --scope "dag-recall" --budget 4000 --ttl 30 +# +# Grant Issued +# ───────────────────────────────────────────── +# Grant ID: g-20260316-003 +# Scope: dag-recall +# Token budget: 4,000 +# Expires: 2026-03-16T11:00:00Z (in 30 min) +# Status: active +# +# python3 guard.py --validate g-20260316-003 +# +# ✓ Grant: g-20260316-003 +# Status: active +# Scope: dag-recall +# Tokens remaining: 2,750 / 4,000 +# Expires in: 18.5 min +# +# python3 guard.py --consume g-20260316-003 --tokens 500 +# +# Consumed 500 tokens from g-20260316-003 +# Remaining: 2,250 / 4,000 +# +# python3 guard.py --stats +# +# Expansion Grant Statistics +# ────────────────────────────────────────────── +# Active grants: 2 +# Total issued: 12 +# Total expired: 6 +# Total revoked: 1 +# Total exhausted: 3 +# Tokens granted: 48,000 +# Tokens consumed: 28,750 +# Utilization: 59.9% diff --git a/skills/openclaw-native/expansion-grant-guard/guard.py b/skills/openclaw-native/expansion-grant-guard/guard.py new file mode 100755 index 0000000..191d4f0 --- /dev/null +++ b/skills/openclaw-native/expansion-grant-guard/guard.py @@ -0,0 +1,504 @@ +#!/usr/bin/env python3 +"""Expansion Grant Guard — YAML-based delegation grant ledger. + +Issues, validates, and tracks scoped permission grants for sub-agent +expansions with token budgets and auto-expiry. + +Usage: + python3 guard.py --issue --scope "dag-recall" --budget 4000 --ttl 30 + python3 guard.py --validate + python3 guard.py --consume --tokens 500 + python3 guard.py --revoke + python3 guard.py --list + python3 guard.py --sweep + python3 guard.py --audit + python3 guard.py --stats + python3 guard.py --status + python3 guard.py --format json +""" + +import argparse +import json +import os +import re +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +# ── Paths ──────────────────────────────────────────────────────────────────── + +OPENCLAW_DIR = Path.home() / ".openclaw" +STATE_DIR = OPENCLAW_DIR / "skill-state" / "expansion-grant-guard" +LEDGER_PATH = STATE_DIR / "ledger.yaml" +STATE_PATH = STATE_DIR / "state.yaml" + +DEFAULT_TTL_MINUTES = 30 +MAX_TTL_MINUTES = 1440 # 24 hours +DEFAULT_BUDGET = 4000 +MAX_HISTORY = 50 + + +# ── YAML-lite read/write ──────────────────────────────────────────────────── + +def load_ledger(): + """Load the grant ledger from YAML-like file.""" + if not LEDGER_PATH.exists(): + return {"grants": [], "counter": 0} + try: + text = LEDGER_PATH.read_text() + data = json.loads(text) # stored as JSON for reliability + return data + except Exception: + return {"grants": [], "counter": 0} + + +def save_ledger(ledger): + """Save ledger to disk.""" + STATE_DIR.mkdir(parents=True, exist_ok=True) + LEDGER_PATH.write_text(json.dumps(ledger, indent=2, default=str)) + + +def load_state(): + """Load aggregate state.""" + if STATE_PATH.exists(): + try: + return json.loads(STATE_PATH.read_text()) + except Exception: + pass + return { + "total_issued": 0, + "total_expired": 0, + "total_revoked": 0, + "total_exhausted": 0, + "total_tokens_granted": 0, + "total_tokens_consumed": 0, + "grant_history": [], + } + + +def save_state(state): + STATE_DIR.mkdir(parents=True, exist_ok=True) + # Trim history + if len(state.get("grant_history", [])) > MAX_HISTORY: + state["grant_history"] = state["grant_history"][-MAX_HISTORY:] + STATE_PATH.write_text(json.dumps(state, indent=2, default=str)) + + +# ── Grant operations ───────────────────────────────────────────────────────── + +def generate_grant_id(ledger): + """Generate a sequential grant ID for today.""" + counter = ledger.get("counter", 0) + 1 + ledger["counter"] = counter + today = datetime.now(timezone.utc).strftime("%Y%m%d") + return f"g-{today}-{counter:03d}" + + +def now_utc(): + return datetime.now(timezone.utc) + + +def parse_iso(s): + """Parse ISO datetime string.""" + try: + return datetime.fromisoformat(s.replace("Z", "+00:00")) + except Exception: + return None + + +def is_expired(grant): + """Check if a grant has expired.""" + expires = parse_iso(grant.get("expires_at", "")) + if expires and now_utc() > expires: + return True + return False + + +def is_exhausted(grant): + """Check if token budget is consumed.""" + return grant.get("tokens_consumed", 0) >= grant.get("token_budget", 0) + + +def effective_status(grant): + """Compute effective status considering time and budget.""" + if grant.get("status") in ("revoked",): + return "revoked" + if is_expired(grant): + return "expired" + if is_exhausted(grant): + return "exhausted" + return grant.get("status", "active") + + +# ── Commands ───────────────────────────────────────────────────────────────── + +def cmd_issue(args): + """Issue a new grant.""" + ledger = load_ledger() + state = load_state() + + ttl = min(args.ttl, MAX_TTL_MINUTES) + budget = args.budget + + grant_id = generate_grant_id(ledger) + issued = now_utc() + expires = issued + timedelta(minutes=ttl) + + grant = { + "grant_id": grant_id, + "scope": args.scope, + "issued_at": issued.isoformat(), + "expires_at": expires.isoformat(), + "token_budget": budget, + "tokens_consumed": 0, + "status": "active", + "issuer": args.issuer or "parent-session", + "metadata": {}, + } + if args.reason: + grant["metadata"]["reason"] = args.reason + + ledger["grants"].append(grant) + save_ledger(ledger) + + state["total_issued"] = state.get("total_issued", 0) + 1 + state["total_tokens_granted"] = state.get("total_tokens_granted", 0) + budget + save_state(state) + + if args.format == "json": + print(json.dumps(grant, indent=2)) + else: + print("Grant Issued") + print("─" * 50) + print(f" Grant ID: {grant_id}") + print(f" Scope: {args.scope}") + print(f" Token budget: {budget:,}") + print(f" Expires: {expires.isoformat()} (in {ttl} min)") + print(f" Status: active") + return 0 + + +def cmd_validate(args): + """Validate a grant — check active, not expired, budget remaining.""" + ledger = load_ledger() + grant = None + for g in ledger["grants"]: + if g["grant_id"] == args.validate: + grant = g + break + + if not grant: + print(f"Grant not found: {args.validate}") + return 1 + + status = effective_status(grant) + remaining = grant["token_budget"] - grant["tokens_consumed"] + expires = parse_iso(grant["expires_at"]) + ttl_remaining = (expires - now_utc()).total_seconds() / 60 if expires else 0 + + valid = status == "active" + + if args.format == "json": + print(json.dumps({ + "grant_id": grant["grant_id"], + "valid": valid, + "status": status, + "scope": grant["scope"], + "tokens_remaining": max(0, remaining), + "minutes_remaining": max(0, round(ttl_remaining, 1)), + }, indent=2)) + else: + icon = "✓" if valid else "✗" + print(f"{icon} Grant: {grant['grant_id']}") + print(f" Status: {status}") + print(f" Scope: {grant['scope']}") + print(f" Tokens remaining: {max(0, remaining):,} / {grant['token_budget']:,}") + if ttl_remaining > 0: + print(f" Expires in: {max(0, round(ttl_remaining, 1))} min") + else: + print(f" Expired: {grant['expires_at']}") + + return 0 if valid else 1 + + +def cmd_consume(args): + """Record token consumption against a grant.""" + ledger = load_ledger() + state = load_state() + + grant = None + for g in ledger["grants"]: + if g["grant_id"] == args.consume: + grant = g + break + + if not grant: + print(f"Grant not found: {args.consume}") + return 1 + + status = effective_status(grant) + if status != "active": + print(f"Grant {args.consume} is {status} — cannot consume") + return 1 + + remaining = grant["token_budget"] - grant["tokens_consumed"] + if args.tokens > remaining: + print(f"Insufficient budget: requested {args.tokens}, remaining {remaining}") + return 1 + + grant["tokens_consumed"] += args.tokens + state["total_tokens_consumed"] = state.get("total_tokens_consumed", 0) + args.tokens + + # Check if now exhausted + if grant["tokens_consumed"] >= grant["token_budget"]: + grant["status"] = "exhausted" + state["total_exhausted"] = state.get("total_exhausted", 0) + 1 + close_grant(state, grant, "exhausted") + + save_ledger(ledger) + save_state(state) + + new_remaining = grant["token_budget"] - grant["tokens_consumed"] + if args.format == "json": + print(json.dumps({ + "grant_id": grant["grant_id"], + "tokens_consumed": args.tokens, + "tokens_remaining": new_remaining, + "status": effective_status(grant), + }, indent=2)) + else: + print(f"Consumed {args.tokens:,} tokens from {grant['grant_id']}") + print(f" Remaining: {new_remaining:,} / {grant['token_budget']:,}") + if new_remaining == 0: + print(f" Status: exhausted") + return 0 + + +def cmd_revoke(args): + """Revoke a grant early.""" + ledger = load_ledger() + state = load_state() + + grant = None + for g in ledger["grants"]: + if g["grant_id"] == args.revoke: + grant = g + break + + if not grant: + print(f"Grant not found: {args.revoke}") + return 1 + + if grant["status"] == "revoked": + print(f"Grant {args.revoke} is already revoked") + return 0 + + grant["status"] = "revoked" + state["total_revoked"] = state.get("total_revoked", 0) + 1 + close_grant(state, grant, "revoked") + + save_ledger(ledger) + save_state(state) + + if args.format == "json": + print(json.dumps({"grant_id": grant["grant_id"], "status": "revoked"}, indent=2)) + else: + print(f"Revoked grant: {grant['grant_id']}") + return 0 + + +def cmd_list(args): + """List active grants.""" + ledger = load_ledger() + active = [g for g in ledger["grants"] if effective_status(g) == "active"] + + if args.format == "json": + print(json.dumps(active, indent=2)) + else: + print(f"Active Grants: {len(active)}") + print("─" * 60) + if not active: + print(" No active grants") + for g in active: + remaining = g["token_budget"] - g["tokens_consumed"] + expires = parse_iso(g["expires_at"]) + ttl = (expires - now_utc()).total_seconds() / 60 if expires else 0 + print(f" {g['grant_id']} scope={g['scope']} " + f"tokens={remaining:,}/{g['token_budget']:,} " + f"expires={max(0, round(ttl))}min") + return 0 + + +def cmd_sweep(args): + """Clean up expired grants.""" + ledger = load_ledger() + state = load_state() + swept = 0 + + for g in ledger["grants"]: + if g["status"] == "active" and is_expired(g): + g["status"] = "expired" + state["total_expired"] = state.get("total_expired", 0) + 1 + close_grant(state, g, "expired") + swept += 1 + + save_ledger(ledger) + save_state(state) + + if args.format == "json": + print(json.dumps({"swept": swept}, indent=2)) + else: + print(f"Sweep complete: {swept} grants expired") + return 0 + + +def cmd_audit(args): + """Show full audit log.""" + state = load_state() + history = state.get("grant_history", []) + + if args.format == "json": + print(json.dumps(history, indent=2)) + else: + print(f"Grant Audit Log: {len(history)} entries") + print("─" * 65) + for h in reversed(history[-20:]): + print(f" {h['grant_id']} scope={h['scope']} " + f"status={h['final_status']} " + f"tokens={h['tokens_consumed']}/{h['token_budget']} " + f"closed={h.get('closed_at', '?')[:16]}") + return 0 + + +def cmd_stats(args): + """Show grant statistics.""" + state = load_state() + ledger = load_ledger() + active_count = sum(1 for g in ledger["grants"] if effective_status(g) == "active") + + if args.format == "json": + stats = { + "active_grants": active_count, + "total_issued": state.get("total_issued", 0), + "total_expired": state.get("total_expired", 0), + "total_revoked": state.get("total_revoked", 0), + "total_exhausted": state.get("total_exhausted", 0), + "total_tokens_granted": state.get("total_tokens_granted", 0), + "total_tokens_consumed": state.get("total_tokens_consumed", 0), + } + print(json.dumps(stats, indent=2)) + else: + print("Expansion Grant Statistics") + print("─" * 50) + print(f" Active grants: {active_count}") + print(f" Total issued: {state.get('total_issued', 0)}") + print(f" Total expired: {state.get('total_expired', 0)}") + print(f" Total revoked: {state.get('total_revoked', 0)}") + print(f" Total exhausted: {state.get('total_exhausted', 0)}") + tg = state.get('total_tokens_granted', 0) + tc = state.get('total_tokens_consumed', 0) + print(f" Tokens granted: {tg:,}") + print(f" Tokens consumed: {tc:,}") + if tg > 0: + pct = round(tc / tg * 100, 1) + print(f" Utilization: {pct}%") + return 0 + + +def cmd_status(args): + """Show current status summary.""" + state = load_state() + ledger = load_ledger() + active = [g for g in ledger["grants"] if effective_status(g) == "active"] + + if args.format == "json": + print(json.dumps({ + "active_grants": len(active), + "total_issued": state.get("total_issued", 0), + "total_tokens_consumed": state.get("total_tokens_consumed", 0), + }, indent=2)) + else: + print("Expansion Grant Guard — Status") + print("─" * 50) + print(f" Active grants: {len(active)}") + print(f" Total issued: {state.get('total_issued', 0)}") + print(f" Tokens consumed: {state.get('total_tokens_consumed', 0):,}") + if active: + print() + for g in active[:5]: + remaining = g["token_budget"] - g["tokens_consumed"] + print(f" {g['grant_id']} {g['scope']} {remaining:,} tokens remaining") + return 0 + + +# ── Helpers ────────────────────────────────────────────────────────────────── + +def close_grant(state, grant, final_status): + """Record a closed grant in history.""" + entry = { + "grant_id": grant["grant_id"], + "scope": grant["scope"], + "issued_at": grant["issued_at"], + "closed_at": now_utc().isoformat(), + "token_budget": grant["token_budget"], + "tokens_consumed": grant["tokens_consumed"], + "final_status": final_status, + } + history = state.get("grant_history", []) + history.append(entry) + state["grant_history"] = history + + +# ── Main ───────────────────────────────────────────────────────────────────── + +def main(): + parser = argparse.ArgumentParser( + description="Expansion Grant Guard — delegation grant ledger" + ) + parser.add_argument("--issue", action="store_true", help="Issue a new grant") + parser.add_argument("--validate", type=str, help="Validate a grant by ID") + parser.add_argument("--consume", type=str, help="Consume tokens from a grant") + parser.add_argument("--revoke", type=str, help="Revoke a grant") + parser.add_argument("--list", action="store_true", help="List active grants") + parser.add_argument("--sweep", action="store_true", help="Clean up expired grants") + parser.add_argument("--audit", action="store_true", help="Show audit log") + parser.add_argument("--stats", action="store_true", help="Grant statistics") + parser.add_argument("--status", action="store_true", help="Current status") + # Issue options + parser.add_argument("--scope", type=str, default="general", help="Grant scope") + parser.add_argument("--budget", type=int, default=DEFAULT_BUDGET, help="Token budget") + parser.add_argument("--ttl", type=int, default=DEFAULT_TTL_MINUTES, help="TTL in minutes") + parser.add_argument("--issuer", type=str, help="Grant issuer identifier") + parser.add_argument("--reason", type=str, help="Reason for grant") + # Consume options + parser.add_argument("--tokens", type=int, default=0, help="Tokens to consume") + # Output + parser.add_argument("--format", choices=["text", "json"], default="text") + + args = parser.parse_args() + + if args.issue: + return cmd_issue(args) + elif args.validate: + return cmd_validate(args) + elif args.consume: + return cmd_consume(args) + elif args.revoke: + return cmd_revoke(args) + elif args.list: + return cmd_list(args) + elif args.sweep: + return cmd_sweep(args) + elif args.audit: + return cmd_audit(args) + elif args.stats: + return cmd_stats(args) + elif args.status: + return cmd_status(args) + else: + parser.print_help() + return 1 + + +if __name__ == "__main__": + sys.exit(main())