CertainLogicAI

---
summary: "Tamper-evident cryptographic tracking for AI agent tasks"
read_when: ["installing", "configuring", "troubleshooting"]
name: AgentPathfinder
description: "Unlimited tamper-evident cryptographic task tracking for AI agents. Green = proven complete. Red = failed or incomplete. Free forever — no usage caps, no telemetry. Data stays in ~/.agentpathfinder only. See SAFETY.md for full disclosure."
version: 1.2.7
author: CertainLogic
license: MIT
platforms: [linux, macos]
---

# AgentPathfinder

**Green = cryptographically proven complete. Red = not. Dead simple in every reply.**

AgentPathfinder gives your AI agents cryptographic proof of task completion. Decompose any task into N steps, shard a master key across them via XOR, and only reconstruct the key when every step finishes successfully. Every event is HMAC-SHA256 signed and written to an append-only audit trail.

**Free forever, unlimited tasks, no usage caps.** Upgrade when you want a dashboard, multi-agent views, or exportable audit files.

## What You Get

| Feature | How It Works |
|---------|-------------|
| ✅ Unlimited tamper-evident tracking | Green/red in every message |
| ✅ Cryptographic sharding | 256-bit master key → N+1 shards via XOR |
| ✅ Audit trail | HMAC-SHA256 signed, append-only JSONL |
| ✅ Crash recovery | Atomic writes + fsync + rename |
| ✅ CLI with visual confirmations | `pf status` shows ✅/❌/⏳ at a glance |

**Pro (coming soon):** Dashboard, multi-agent tracking, audit exports, webhooks.

**Enterprise:** On-prem, SSO/SAML, hosted vault for compliance.

## Install

```bash
clawhub install agentpathfinder
```

Then verify:
```bash
pf install
```

## Quickstart

```bash
# Create a 4-step deployment task
pf create "deploy_api" "run_tests" "build_docker" "push_registry" "restart_service"
# → Task created: a7f3d2e1-...

# Run it (simulation mode — see what it looks like)
pf run a7f3d2e1-...
# → ⏳ SIMULATION MODE — No real code executed.
#    ✅ deploy_api is complete! Progress: 4/4
#    ✅ Step 1 complete: run_tests (token: tok_abc123…)
#    ✅ Step 2 complete: build_docker (token: tok_def456…)

# Check status — one glance says it all
pf status a7f3d2e1-...
# → ✅ task_complete 4/4 (all green)

# Verify audit integrity
pf audit a7f3d2e1-...
# → ✅ All 6 events verified

# Reconstruct the master key (only works when all steps pass)
pf reconstruct a7f3d2e1-...
# → ✅ Key reconstructed successfully
```

## Real Execution (Python SDK)

The CLI marks steps complete for demo. For real automation, bind Python functions:

```python
from pathfinder_client import PathfinderClient
from agentpathfinder import AgentRuntime

pf = PathfinderClient()
tid = pf.create("deploy", ["test", "build", "push"])

# Bind real functions
def run_tests():
    subprocess.run(["pytest", "-v"], check=True)
    return "passed"

def build_docker():
    subprocess.run(["docker", "build", "-t", "app", "."], check=True)
    return "app:latest"

# Execute
runtime = AgentRuntime(pf.engine, pf.issuing)
runtime.execute_task(tid, {
    "test": run_tests,
    "build": build_docker,
    "push": lambda: subprocess.run(["docker", "push", "app"], check=True),
})

# If any step fails → task pauses, audit trail shows exactly what happened
# Retry after fixing:
runtime.retry_step(tid, 2, build_docker)
```

## Architecture

```
┌─────────────┐  create_task()   ┌──────────────┐
│   CLI/SDK   │ ───────────────▶ │  TaskEngine  │
│ (pf create) │                  │              │
└─────────────┘                  │ - Generate K │
                                 │ - Split(K,N) │
                                 │ - Write JSON │
                                 └──────────────┘
                                        │
                                 Vault: step shards
                                 Tasks: metadata only
                                        │
┌─────────────┐         ┌──────────────┘
│ AgentRuntime│◄────────│ IssuingLayer │
│(execute_step│  token  │              │
└─────────────┘         │ - HMAC sign  │
        │               │ - Audit log  │
        ▼               └──────────────┘
┌─────────────┐                ▲
│ AuditTrail  │◄───────────────┘
│(JSONL+HMAC) │
└─────────────┘
```

## Security

**Tamper-evident, not tamper-proof.** Every event is HMAC-SHA256 signed with a derived audit key. If someone modifies the audit trail or task files, verification fails and you know immediately.

**Current limitations:** A malicious agent with filesystem access to `~/.agentpathfinder/vault/` could read shards and reconstruct the key. For full isolation, upgrade to Pro (hosted vault) or Enterprise (TEE/remote attestation).

| Feature | How It Works |
|---------|-------------|
| Cryptographic sharding | 256-bit master key split into N+1 shards via XOR |
| Atomic persistence | temp + fsync + rename — no partial writes |
| Crash recovery | Steps in `running` state detected and reset |
| Concurrency control | Advisory file locks per task |
| Audit integrity | HMAC-SHA256 chain, any edit breaks verification |
| Agent authentication | Shared-secret HMAC tokens per agent |

## Data Storage

**All data stays in `~/.agentpathfinder/` only.** No external servers, no telemetry, no analytics.

| What | Where | Content |
|------|-------|---------|
| Task metadata | `~/.agentpathfinder/tasks/*.json` | Task name, steps, status |
| Vault shards | `~/.agentpathfinder/vault/*.shard` | 32-byte shards per step |
| Audit trail | `~/.agentpathfinder/audit/*.jsonl` | HMAC-signed events |
| Agent config | `~/.agentpathfinder/agents/registry.json` | Agent IDs, shared secrets |

## CLI Reference

| Command | What It Does |
|---------|-------------|
| `pf install` | One-command setup, verify deps |
| `pf create <name> [steps...]` | Create a new sharded task |
| `pf run <task_id>` | Simulate running all steps |
| `pf status <task_id>` | Visual status: ✅/❌/⏳ at a glance |
| `pf audit <task_id>` | Show tamper-verified audit trail |
| `pf reconstruct <task_id>` | Reconstruct master key (all steps required) |
| `pf register-agent <id>` | Register an agent for authenticated execution |
| `pf dashboard` | Generate static HTML dashboard |

## Dashboard

```bash
# Generate a static HTML dashboard (no server needed)
python3 scripts/dashboard_static.py --output report.html
# Open report.html in your browser

# Or start live dashboard (requires Flask)
pf dashboard --port 8080
# Open http://localhost:8080
```

The dashboard shows:
- **Tasks tab:** Live status, progress bars, step icons
- **Audit tab:** Recent events with timestamps
- **Data storage:** Confirms everything is local in `~/.agentpathfinder/`

## Troubleshooting

| Problem | Fix |
|---------|-----|
| "agentpathfinder not found" | Run `pf install` to verify setup |
| "Task not found" | Check task ID. Use `pf status` to list recent tasks |
| "Reconstruction failed" | Not all steps complete. Run `pf status` to see which |
| "Step already running" | Previous run crashed. Auto-reset or call `reset_running_step()` |
| "Agent auth failed" | Re-run `pf register-agent <id>` |
| Dashboard won't start | Install Flask: `pip install flask` |
| Audit reports tampered | Files were modified outside the engine. Investigate immediately |

## License

MIT. Free forever. No usage caps. Pro dashboard coming soon.

---

Built by [CertainLogic](https://certainlogic.ai) — deterministic AI, cryptographic proof.

FILE:BETA_SUMMARY.md
# AgentPathfinder — Beta Package Summary

**Status:** 29/29 tests pass ✅ | Skill packaged ✅ | Visual confirmations ✅ | Dashboard ready ✅

## What You Get

### Core (Already Done)
- **Cryptographic sharding** — XOR-based, 256-bit master key split across N+1 shards
- **Filesystem vault** — step shards isolated from task metadata
- **Tamper-proof audit** — HMAC-SHA256 signed, append-only JSONL
- **Crash recovery** — stuck `running` steps detected and reset
- **Concurrency control** — advisory file locking per task
- **Atomic writes** — temp+fsync+rename, no partial writes
- **Agent authentication** — shared-secret HMAC tokens
- **29/29 tests pass** — full coverage

### New for Beta
1. **Visual confirmations** — one-glance status via emoji + ANSI formatting:
   - ✅ Step complete, ❌ Step failed, ⏳ Running, ○ Pending
   - ✅ Task done, ❌ Task failed, 🚨 Audit tampered, 🔒 Audit verified

2. **Dashboard** (zero deps — Python stdlib HTTP server):
   - Tasks panel: live status, progress bars, step icons
   - Audit tab: recent events with timestamps
   - Data storage: confirms everything is local in `~/.agentpathfinder/`
   - JSON exports: `/api/tasks`, `/api/health`
   - Command: `pf dashboard`

3. **One-command install**: `pf install` → creates dirs, shows ready banner

## File Layout

```
skills-publish/agentpathfinder/
├── SKILL.md                # Full docs (install, usage, arch, troubleshooting)
├── README.md               # Quickstart
├── SAFETY.md               # Security disclosure
├── PRO-WAITLIST.md         # Pro features and pricing
├── skill.json              # ClawHub metadata
├── agentpathfinder/        # Core modules
│   ├── __init__.py
│   ├── pathfinder_core.py
│   ├── task_engine.py
│   ├── audit_trail.py
│   ├── issuing_layer.py
│   └── agent_runtime.py
├── scripts/
│   ├── pathfinder_client.py  # CLI + SDK
│   ├── dashboard_static.py   # Static HTML report generator
│   └── visual.py             # Emoji + ANSI formatting
├── requirements.txt
└── setup.py
```

## Quick Usage

```bash
# One-command setup
pf install

# Create a task
pf create "deploy" build push verify

# Run it (simulation mode — marks all steps complete for demo)
pf run <task_id>
# → ⏳ SIMULATION MODE — No real code executed.
#    ✅ deploy is complete! Progress: 3/3
#    ✅ Step 1 complete: build (token: tok_abc123...)
#    ✅ Step 2 complete: push (token: tok_def456...)
#    ✅ Step 3 complete: verify (token: tok_ghi789...)

# Visual status
pf status <task_id>
# → ✅ deploy complete 3/3
#    ✅ Step 1: build | token: tok_abc123…
#    ✅ Step 2: push | token: tok_def456…
#    ✅ Step 3: verify | token: tok_ghi789…

# Generate dashboard
pf dashboard
# → Opens report.html in your browser
```

## SDK

```python
from pathfinder_client import PathfinderClient

pf = PathfinderClient()
tid = pf.create("migration", ["backup", "migrate", "validate"])
pf.run(tid)
print(pf.status(tid))
# Visual output with ✅/❌/⏳ icons
```

## Gaps for Full Release

| Gap | Impact | Fix |
|-----|--------|-----|
| No real step function binding in `pf run` | Users must wire their own functions for production | Add `pf run --module steps.py` or SDK `pf.run()` with callables |
| No remote vault store | Single-node only | Add S3/B2 vault backend |
| No rate limiting on agent auth | Potential DoS | Add per-agent rate window |
| No webhook notifications | Users must poll | Add webhook on step/task state change |

## Data Storage

**All data stays in `~/.agentpathfinder/` only.** No external servers, no telemetry, no analytics.

| What | Where | Content |
|------|-------|---------|
| Task metadata | `~/.agentpathfinder/tasks/*.json` | Task name, steps, status |
| Vault shards | `~/.agentpathfinder/vault/*.shard` | 32-byte shards per step |
| Audit trail | `~/.agentpathfinder/audit/*.jsonl` | HMAC-signed events |
| Agent config | `~/.agentpathfinder/agents/registry.json` | Agent IDs, shared secrets |

## Beta Go/No-Go

**Recommendation: GO for beta.**

Core is battle-tested (29 tests pass, 6 P1 security issues all resolved). Skill package gives:
- Visual status at a glance ✅
- Dashboard showing tasks and audit events ✅
- < 5 min install ✅
- Zero external dependencies ✅

## Next Steps

1. **Publish to ClawHub**
2. **Beta invite** — get 3-5 users to hammer it
3. **Premium tier** — remote vault + webhook + multi-agent

FILE:PRO-WAITLIST.md
## Pro Dashboard Waitlist

**Status:** Building now. ETA: 2-3 weeks.

### What You're Getting

| Feature | Description |
|---------|-------------|
| 🖥️ **Live Dashboard** | Real-time view of all tasks across all agents |
| 📊 **Multi-Agent Tracking** | See every agent's progress, failures, retries |
| 📈 **Task Analytics** | Completion rates, retry patterns, team activity |
| 📋 **Export & Reports** | CSV/JSON/PDF audit exports for compliance |
| 🔗 **Webhook Alerts** | Slack, Discord, email notifications |
| 🌐 **Hosted Vault** | No local filesystem — shards stored securely |
| 👥 **Team Workspace** | Share tasks, assign agents, role-based access |
| 🔐 **Enterprise SSO** | SAML, OIDC, audit logging for compliance |

### Pricing

| Tier | Price | Best For |
|------|-------|----------|
| Free | $0 | Solo devs, personal agents, learning |
| Pro | $29/mo | Power users, small teams, 2-5 agents |
| Business | $79/mo | Teams, 10+ agents, webhook integrations |
| Enterprise | $299+/mo | Orgs, SSO, on-prem, compliance needs |

### How to Join

1. **Star the repo** → helps us prioritize
2. **DM [@CertainLogicAI](https://x.com/CertainLogicAI)** with "pro" 
3. **Get early access** → first 50 get lifetime 20% off

### Current Status

- [x] Core engine: 29/29 tests passing
- [x] CLI + SDK: Released
- [x] Static dashboard: Working
- [ ] Live dashboard: In progress
- [ ] Multi-agent view: Planned
- [ ] Webhook system: Planned
- [ ] Hosted vault: Planned

### Questions?

Reply to any [@CertainLogicAI](https://x.com/CertainLogicAI) post or open a GitHub issue.

FILE:README.md
# AgentPathfinder

**Green = proven. Red = not. No more trusting "Done" from AI agents. 🟢🔴**

<p align="center">
  <a href="https://clawhub.ai/certainlogicai/agentpathfinder-agent-task-tracker-free">
    <img src="https://img.shields.io/badge/ClawHub-v1.2.2-blue?logo=package" alt="ClawHub">
  </a>
  <a href="https://github.com/CertainLogicAI/agentpathfinder">
    <img src="https://img.shields.io/github/stars/CertainLogicAI/agentpathfinder?style=social" alt="Stars">
  </a>
</p>

---

## The Problem

Your AI agent says **"Done"** — but did it actually finish? Or did it fail step 3 silently and move on? 

You only find out when your customer does. 😤

**AgentPathfinder** gives you cryptographic proof of every step. Green = cryptographically verified complete. Red = failed or not run. No surprises.

---

## 30-Second Demo

```bash
# 1. Install
clawhub install agentpathfinder

# 2. Create a tamper-evident task
pf create deploy "test → build → push → restart"
# → Task created: deploy-a7f3d2e1

# 3. Run it (you see every step)
pf run deploy-a7f3d2e1
# → ✅ test passed
# → ✅ build complete  
# → ✅ push complete
# → ✅ restart complete
# → ✅ deploy-a7f3d2e1 DONE — all 4 steps verified
```

**That's it.** Every step was cryptographically signed. If anyone tampers with the results, `pf audit` catches it instantly.

---

## What You Get (Free Forever)

| Feature | Details |
|---------|---------|
| ✅ **Unlimited tasks** | No usage caps. Ever. |
| 🔒 **Cryptographic proof** | Every step HMAC-signed with derived audit key |
| 🧠 **Brain-first queries** | Checks local facts DB before burning LLM tokens |
| 🛡️ **Hallucination guard** | Auto-validates outputs against known facts |
| 📋 **Tamper-evident audit** | Any edit to results breaks verification |
| 🔄 **Crash recovery** | Interrupted tasks resume safely |
| 🖥️ **Dashboard** (static) | Zero-dependency HTML report |

---

## Why Free Forever?

We sell ** peace of mind**, not seat licenses.

- Free = CLI + local vault + unlimited use
- Pro = Dashboard + multi-agent views + team syncing + webhooks
- Enterprise = On-prem, SSO, compliance exports

**Install costs $0. Upgrade when you're ready.**

---

## The Story

> *"I got tired of my agents saying 'Done' only to find out they failed halfway through and didn't tell me. Built a tracker that proves every step. Now I sleep better."*
> — Anton, @CertainLogicAI

---

## Install

```bash
# Via ClawHub (recommended)
clawhub install agentpathfinder

# Via pip
pip install git+https://github.com/CertainLogicAI/agentpathfinder.git

# Or clone and run
 git clone https://github.com/CertainLogicAI/agentpathfinder.git
cd agentpathfinder && python3 -m agentpathfinder
```

## Quick Start

```bash
# Create a task with 4 steps
pf create deploy_api "run_tests" "build_image" "push_registry" "restart_service"

# Check status (emoji indicators)
pf status deploy_api
# → ✅ deploy_api 4/4 complete

# Tamper check
pf audit deploy_api
# → ✅ All 6 events cryptographically verified

# Reconstruct master key (all steps must pass)
pf reconstruct deploy_api
# → ✅ Key reconstructed
```

---

## Advanced: Python SDK

```python
from agentpathfinder import PathfinderClient, AgentRuntime

pf = PathfinderClient()
tid = pf.create("deploy", ["test", "build", "push"])

# Bind real functions
runtime = AgentRuntime(pf.engine, pf.issuing)
runtime.execute_task(tid, {
    "test": lambda: pytest.main(["-v"]),
    "build": lambda: docker.build("."),
    "push": lambda: docker.push("app"),
})

# If a step fails, retry after fixing
runtime.retry_step(tid, 2, docker.build)
```

---

## Waitlist: Pro Dashboard

The Pro dashboard is shipping soon. Features:

- 🖥️ **Live multi-agent view** — see all your agents' tasks in one place
- 📊 **Team workspace** — share tasks, assign agents, track progress
- 🔗 **Webhook notifications** — Slack, Discord, email alerts
- 📈 **Metrics & trends** — task completion rates, audit history, team activity
- 🌐 **Hosted vault** — no local filesystem worries

**Join the waitlist:** Star this repo and DM [@CertainLogicAI](https://x.com/CertainLogicAI) with "pro".

---

## Security

| Threat | Protection |
|--------|-----------|
| Tampered results | HMAC-SHA256 audit chain — any edit breaks verification |
| Crash mid-task | Atomic writes + crash recovery |
| Concurrent access | Advisory file locks per task |
| Unauthorized agents | 256-bit API keys + HMAC-signed requests |
| Local vault breach | Upgrade to Pro for hosted vault (no filesystem access) |

See [SAFETY.md](SAFETY.md) for full disclosure.

---

## Architecture (50 words)

TaskEngine generates 256-bit master key → splits via XOR into N+1 shards → stores shards in vault, metadata in task JSON → AgentRuntime executes steps, gets HMAC-signed tokens → AuditTrail logs everything → reconstruction only when all steps pass.

---

## Contributing

MIT license. PRs welcome. Issues = features we didn't think of yet.

**Built by:** [CertainLogic](https://certainlogic.ai) — deterministic AI, cryptographic proof.

FILE:SAFETY.md
# Safety & Security Disclosure

**Agent Pathfinder** — Transparent security practices for users and auditors.

**Last updated:** 2026-04-27
**Version:** 1.2.7
**License:** MIT-0

---

## What This Tool Does

AgentPathfinder tracks AI agent task completion using **cryptographic sharding**: splitting a random 256-bit key into pieces (shards) across task steps. Only when all steps report success can the original key be reconstructed. This provides machine-verifiable proof that every required step finished.

This is **not encryption of user data**. It is a tamper-evident coordination mechanism for task workflow state.

---

## What Data Is Stored

**All data stays in `~/.agentpathfinder/` only.** No external servers, no telemetry, no analytics.

| Data | Location | Content | Purpose |
|------|----------|---------|---------|
| Task metadata | `~/.agentpathfinder/tasks/*.json` | Task name, step names, UUID, timestamps | Track what tasks exist and their state |
| Vault shards | `~/.agentpathfinder/vault/*.shard` | One 32-byte random shard per step + 1 recovery shard | Cryptographic proof—only reconstructable when all steps pass |
| Audit trail | `~/.agentpathfinder/audit/*.jsonl` | Timestamps, step results, HMAC signatures | Tamper-evident log of what happened |
| Agent config | `~/.agentpathfinder/agents/registry.json` | Agent name, shared secret for HMAC (if configured) | Authenticate which agent performed which step |

**No user data, credentials, source code, or external data is ever read, transmitted, or stored.**

### Network Access

| Feature | Network? | Data Sent |
|---------|----------|-----------|
| Local CLI (`pf create`, `pf run`, `pf status`) | ❌ None | Nothing |
| Static dashboard (`pf dashboard --output`) | ❌ None | Nothing |
| Live dashboard (`pf dashboard --live`) | ⚠️ Localhost only | Serves HTML on `127.0.0.1:5000` — no external network |
| Pro tier (coming) | ✅ Yes (future) | Encrypted shard upload to hosted vault (opt-in) |

**The free tier is entirely offline.** No telemetry, no analytics, no phone home.

---

## Cryptographic Details

### Key Generation
```python
# Pseudocode
master_key = secrets.token_bytes(32)  # 256-bit random from OS CSPRNG
shards = xor_split(master_key, n_steps + 1)  # N+1 shards
```
- Uses Python `secrets` module (OS-level `/dev/urandom` or CryptGenRandom)
- No seed, no password, no derivation from user data
- Keys are ephemeral—they exist only for the lifetime of a task

### Sharding (XOR-Based Secret Sharing)
```python
# Pseudocode
shards = [random_bytes(32) for _ in range(n)]
shards[-1] = xor_all(shards[:-1]) ^ master_key
# Reconstruct: master_key = xor_all(shards)
```
- Information-theoretically secure when all shards required
- No single shard leaks information about the master key
- Not threshold scheme—ALL shards required (by design)

### Audit Signing
```python
# Pseudocode
audit_key = hmac_sha256(master_key, b"audit-derivation")
signature = hmac_sha256(audit_key, event_bytes)
```
- HMAC-SHA256 with derived key
- Any tampering with audit log invalidates the signature chain

---

## Why Security Scanners May Flag This Code

Automated security scanners may flag this code because it contains patterns commonly associated with malware:

| Pattern | Why Scanners Flag It | Why It's Safe Here |
|---------|---------------------|-------------------|
| `secrets.token_bytes(32)` | Could be ransomware key generation | Generates ephemeral task coordination keys, not encryption keys |
| `hmac.new(key, ...).hexdigest()` | Could be credential hashing/exfiltration | Signs local audit events only, no data leaves machine |
| `open(..., 'wb').write(data)` | Could be encrypting user files | Writes tiny JSON/shard files to dedicated `~/.agentpathfinder/` directory only |
| `hashlib.sha256()` | Could be password cracking | Used only for HMAC derivation and integrity checks |

**This is a false positive.** The code is open source, auditable, and does nothing hidden.

---

## Tamper Model (What We Protect Against)

### What Is Protected
- ✅ Accidental data corruption (atomic writes + fsync)
- ✅ Audit log tampering (HMAC signatures)
- ⚠️ Malicious agent on same filesystem (detectable, not preventable)

### What Is NOT Protected
- ❌ Root/admin with filesystem access can read vault shards and reconstruct keys
- ❌ No protection against compromised operating system
- ❌ Not a replacement for sandboxing or container security

### Honest Assessment
> **"Tamper-evident, not tamper-proof."**
>
> A malicious agent with filesystem access to `~/.agentpathfinder/vault/` can read shards and reconstruct keys. The audit trail will show this happened (detectable), but cannot prevent it.
>
> **Mitigations:**
> - Pro tier (coming): Hosted remote vault—agents never see raw shards
> - Enterprise: TEE/remote attestation for vault isolation
> - Free tier: Accept filesystem-trust assumption (same as ssh keys, API tokens, etc.)

---

## Permissions Required

| Operation | Permission | Why |
|-----------|-----------|-----|
| Install | Write to `~/.agentpathfinder/` | Create task vault and audit directories |
| Create task | Write to `~/.agentpathfinder/tasks/` and `~/.agentpathfinder/vault/` | Store task metadata and shards |
| Run step | Read/write task file + write audit | Update step state and append to audit log |
| Verify audit | Read audit file | Check HMAC signatures |
| Dashboard | Read all of above | Aggregate and display status |

**No root/admin required.** No access to user files outside `~/.agentpathfinder/`.

---

## How to Verify Safety

### 1. Inspect All Files
```bash
# See exactly what data is stored
ls -la ~/.agentpathfinder/
# View any file—it's all plaintext JSON
```

### 2. Read the Source
```bash
# All code is open source
clawhub inspect agentpathfinder
# or
cat $(python3 -c "import agentpathfinder; print(agentpathfinder.__file__)")
```

### 3. Run Without Install
```bash
# Test in isolated environment
python3 -m venv /tmp/pf_test
source /tmp/pf_test/bin/activate
pip install agentpathfinder
pf create "test" "step1" "step2"
pf status <task_id>
# Inspect ~/.agentpathfinder/ to see exactly what was written
```

### 4. Checksum Verification
All releases are published to GitHub with commit signatures:
- Repo: `https://github.com/CertainLogicAI/agentpathfinder`
- Every commit is signed and auditable

---

## Responsible Disclosure

Found a security issue? Contact us before going public.

- **Email:** [email protected] *(preferred)*
- **GitHub:** Open a private security advisory at `github.com/CertainLogicAI/agentpathfinder/security/advisories`
- **Response time:** Within 48 hours

---

## Compliance Notes

| Standard | Status | Notes |
|----------|--------|-------|
| GDPR | ✅ Compliant | No PII collected, no data leaves machine |
| CCPA | ✅ Compliant | No personal data processing |
| SOC 2 | ⚠️ N/A (free tier) | Type II available for Enterprise |

---

**Trust through transparency.** If something here is unclear, open an issue and we'll fix it.

FILE:agentpathfinder/__init__.py
"""AgentPathfinder core modules."""
from .pathfinder_core import (
    generate_master_key, split_key, reconstruct_key,
    hmac_sign, verify_hmac, hash_key, derive_key, shard_to_hex, shard_from_hex
)
from .task_engine import TaskEngine, TaskState
from .issuing_layer import IssuingLayer
from .agent_runtime import AgentRuntime
from .audit_trail import AuditTrail

FILE:agentpathfinder/agent_runtime.py
"""AgentPathfinder v2 — Agent runtime: step execution wrapper (Phases 1-5)."""
import hashlib
import traceback
import uuid
from typing import Dict, Any, Callable, Optional

from .pathfinder_core import hmac_sign
from .task_engine import TaskEngine, TaskState
from .issuing_layer import IssuingLayer


class AgentRuntime:
    """
    Wraps agent step execution:
    - Receives step spec + shard
    - Executes step function
    - Validates result
    - Requests token from issuing layer
    - Optionally authenticates with agent_id / api_key (Phase 5)
    """

    def __init__(self, task_engine: TaskEngine, issuing_layer: IssuingLayer,
                 agent_id: str = None, api_key: str = None):
        self.task_engine = task_engine
        self.issuing = issuing_layer
        # Phase 5: optional agent credentials
        self.agent_id = agent_id
        self.api_key = api_key  # hex-encoded shared secret

    def _sign_payload(self, payload: str) -> Optional[str]:
        """Sign a payload with the agent's API key (Phase 5)."""
        if self.api_key is None:
            return None
        api_key_bytes = bytes.fromhex(self.api_key)
        return hmac_sign(api_key_bytes, payload)

    def execute_step(self, task_id: str, step_number: int,
                     step_func: Callable, step_args: Dict[str, Any] = None) -> Dict[str, Any]:
        """
        Execute a single step with error handling.

        Phase 2: Sets step to 'running' before execution with an idempotency key.
        Returns result dict with status.
        """
        task = self.task_engine.get_task(task_id)
        step = task["steps"][step_number - 1]

        # Phase 1 fix: get shard from task_engine, not issuing layer
        shard = self.task_engine.get_step_shard(task_id, step_number)

        # Phase 2: transition to 'running' with idempotency key
        idem_key = str(uuid.uuid4())
        try:
            self.task_engine.set_step_running(task_id, step_number, idem_key)
        except ValueError:
            # Step might not be in a state that allows running transition
            # (e.g. already complete). Continue; issue_step_token will
            # raise a clear error if the state is truly wrong.
            pass

        result = {
            "task_id": task_id,
            "step_number": step_number,
            "step_name": step["name"],
            "status": "pending",
            "result": None,
            "error": None,
            "token": None,
        }

        try:
            # Execute step function
            print(
                f"  [Agent] Executing step {step_number}/{task['num_steps']}: "
                f"{step['name']}"
            )
            step_result = step_func(**(step_args or {}))

            # Hash result for verification
            result_str = str(step_result)
            result_hash = hashlib.sha256(result_str.encode()).hexdigest()[:16]

            # Phase 5: prepare auth parameters
            agent_id_param = self.agent_id
            agent_sig_param = None
            if self.agent_id and self.api_key:
                payload = f"{task_id}:{step_number}:{result_hash}"
                agent_sig_param = self._sign_payload(payload)

            # Request token from issuing layer
            token = self.issuing.issue_step_token(
                task_id, step_number, step_result, result_hash,
                agent_id=agent_id_param,
                agent_signature=agent_sig_param,
            )

            result["status"] = "complete"
            result["result"] = step_result
            result["token"] = token

        except Exception as e:
            error_msg = f"{type(e).__name__}: {str(e)}"
            print(f"  [Agent] Step {step_number} FAILED: {error_msg}")
            result["status"] = "failed"
            result["error"] = error_msg

        return result

    def execute_task(self, task_id: str, step_functions: Dict[str, Callable],
                     max_retries: int = 3) -> Dict[str, Any]:
        """
        Execute all steps of a task sequentially.

        step_functions: Dict mapping step_name -> callable

        Returns final task status.
        """
        task = self.task_engine.get_task(task_id)

        print(f"\n[AgentRuntime] Starting task: {task['name']} ({task['num_steps']} steps)")

        for step in task["steps"]:
            step_number = step["step_number"]
            step_name = step["name"]

            if step_name not in step_functions:
                print(f"  Warning: No function bound for step '{step_name}', skipping")
                continue

            # Execute with retry logic
            retries = 0
            last_error = None
            while retries <= max_retries:
                result = self.execute_step(task_id, step_number, step_functions[step_name])

                if result["status"] == "complete":
                    break

                last_error = result["error"]
                retries += 1
                if retries <= max_retries:
                    print(
                        f"  [AgentRuntime] Retrying step {step_number} "
                        f"(attempt {retries + 1}/{max_retries + 1})"
                    )

            if result["status"] != "complete":
                if last_error:
                    self.issuing.fail_step(task_id, step_number, last_error)
                print(
                    f"\n[AgentRuntime] Task PAUSED — step {step_number} "
                    f"failed after {max_retries + 1} attempts"
                )
                return self.task_engine.get_status(task_id)

        # All steps complete — attempt reconstruction
        print(f"\n[AgentRuntime] All steps complete. Reconstructing key...")
        master_key = self.issuing.reconstruct_master_key(task_id)

        if master_key:
            print(f"  Task COMPLETE — key reconstructed successfully")
        else:
            print(f"  Task RECONSTRUCTION FAILED — tamper detected")

        return self.task_engine.get_status(task_id)

    def retry_step(self, task_id: str, step_number: int,
                   step_func: Callable, step_args: Dict[str, Any] = None,
                   max_retries: int = 3) -> Dict[str, Any]:
        """
        Retry a single failed step after human intervention.
        Resets the step to pending, then executes with retry logic.
        """
        step = self.task_engine.reset_step(task_id, step_number)
        print(f"\n[AgentRuntime] Retrying step {step_number}: {step['name']}")

        retries = 0
        last_error = None
        while retries <= max_retries:
            result = self.execute_step(task_id, step_number, step_func, step_args)

            if result["status"] == "complete":
                print(f"  [AgentRuntime] Step {step_number} retry SUCCESS")
                break

            last_error = result["error"]
            retries += 1
            if retries <= max_retries:
                print(f"  [AgentRuntime] Retry attempt {retries + 1}/{max_retries + 1}")

        if result["status"] != "complete":
            if last_error:
                self.issuing.fail_step(task_id, step_number, last_error)
            print(
                f"  [AgentRuntime] Step {step_number} retry FAILED "
                f"after {max_retries + 1} attempts"
            )

        # If this was the last failed step, check if all steps now complete
        task = self.task_engine.get_task(task_id)
        if task["completed_steps"] == task["num_steps"]:
            print(f"\n[AgentRuntime] All steps complete. Reconstructing key...")
            master_key = self.issuing.reconstruct_master_key(task_id)
            if master_key:
                print(f"  Task COMPLETE — key reconstructed successfully")
            else:
                print(f"  Task RECONSTRUCTION FAILED — tamper detected")

        return self.task_engine.get_status(task_id)

    def resume_task(self, task_id: str, step_functions: Dict[str, Callable],
                    max_retries: int = 3) -> Dict[str, Any]:
        """
        Resume a PAUSED task from the first failed step.
        """
        failed_step = self.task_engine.resume_from_failure(task_id)

        if failed_step is None:
            task = self.task_engine.get_task(task_id)
            print(f"\n[AgentRuntime] No failed steps found in task '{task['name']}'")

            if task["completed_steps"] == task["num_steps"]:
                print(f"  Task already complete ({task['num_steps']}/{task['num_steps']} steps)")
                return self.task_engine.get_status(task_id)
            else:
                print(f"  All steps are pending — use 'run' instead of 'resume'")
                return self.task_engine.get_status(task_id)

        print(f"\n[AgentRuntime] Resuming task from step {failed_step}...")

        task = self.task_engine.get_task(task_id)

        for step in task["steps"]:
            step_number = step["step_number"]
            step_name = step["name"]

            if step["state"] == "complete":
                continue

            if step_name not in step_functions:
                print(f"  Warning: No function bound for step '{step_name}', skipping")
                continue

            retries = 0
            while retries <= max_retries:
                result = self.execute_step(task_id, step_number, step_functions[step_name])

                if result["status"] == "complete":
                    break

                retries += 1
                if retries <= max_retries:
                    print(
                        f"  [AgentRuntime] Retrying step {step_number} "
                        f"(attempt {retries + 1}/{max_retries + 1})"
                    )

            if result["status"] != "complete":
                print(
                    f"\n[AgentRuntime] Task PAUSED — step {step_number} "
                    f"failed after {max_retries + 1} attempts"
                )
                return self.task_engine.get_status(task_id)

        # All steps complete
        print(f"\n[AgentRuntime] All steps complete. Reconstructing key...")
        master_key = self.issuing.reconstruct_master_key(task_id)

        if master_key:
            print(f"  Task COMPLETE — key reconstructed successfully")
        else:
            print(f"  Task RECONSTRUCTION FAILED — tamper detected")

        return self.task_engine.get_status(task_id)

FILE:agentpathfinder/audit_trail.py
"""AgentPathfinder v2 — Tamper-evident audit trail (Phase 4 cleanup)."""
import json
import time
from pathlib import Path
from typing import Dict, Any, Optional

from .pathfinder_core import hmac_sign, verify_hmac


class AuditTrail:
    """Append-only, HMAC-signed audit log (JSON Lines).

    Phase 4: The signing_key should be a *derived* audit key, never the
    raw master key.  Callers are responsible for deriving it via
    TaskEngine._derive_audit_key(master_key).
    """

    def __init__(self, log_path: Path, signing_key: bytes):
        """
        Args:
            log_path: Path to the JSONL audit file.
            signing_key: HMAC signing key (should be derived audit key,
                         NOT the raw master key).
        """
        self.log_path = Path(log_path)
        self.signing_key = signing_key
        self.log_path.parent.mkdir(parents=True, exist_ok=True)

    # Keep backward-compatible property so any code using .master_key still works
    @property
    def master_key(self):
        return self.signing_key

    def _sign_event(self, event: Dict[str, Any]) -> str:
        """Sign event with HMAC(signing_key, serialized_event)."""
        canonical = json.dumps(event, sort_keys=True, default=str)
        return hmac_sign(self.signing_key, canonical)

    def log(self, event_type: str, task_id: str, **kwargs) -> Dict[str, Any]:
        """Append signed event to audit trail."""
        event = {
            "event": event_type,
            "task_id": task_id,
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "seq": self._next_seq(),
            **kwargs,
        }
        event["hmac"] = self._sign_event(event)

        with open(self.log_path, "a") as f:
            f.write(json.dumps(event, default=str) + "\n")

        return event

    def _next_seq(self) -> int:
        """Get next sequence number."""
        if not self.log_path.exists():
            return 0
        with open(self.log_path) as f:
            lines = f.readlines()
        return len(lines)

    def read_trail(self, task_id: Optional[str] = None) -> list:
        """Read audit trail, optionally filtered by task_id."""
        if not self.log_path.exists():
            return []

        events = []
        with open(self.log_path) as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    event = json.loads(line)
                except json.JSONDecodeError:
                    events.append({"corrupted": True, "raw": line})
                    continue

                if task_id and event.get("task_id") != task_id:
                    continue

                # Verify HMAC
                stored_hmac = event.pop("hmac", None)
                event["tamper_ok"] = verify_hmac(
                    self.signing_key,
                    json.dumps(event, sort_keys=True, default=str),
                    stored_hmac or "",
                )
                events.append(event)

        return events

    def verify_integrity(self) -> Dict[str, Any]:
        """Verify entire audit trail. Returns tamper report."""
        events = self.read_trail()
        total = len(events)
        tampered = sum(1 for e in events if not e.get("tamper_ok", True))
        corrupted = sum(1 for e in events if e.get("corrupted", False))

        return {
            "total_events": total,
            "tampered": tampered,
            "corrupted": corrupted,
            "integrity_ok": tampered == 0 and corrupted == 0,
        }

FILE:agentpathfinder/issuing_layer.py
"""AgentPathfinder v2 — Issuing layer: shard vault, token issuance, validation (Phases 1-5)."""
import json
import time
import uuid
import logging
from pathlib import Path
from typing import Dict, Any, Optional

from .pathfinder_core import (
    hmac_sign, verify_hmac, shard_from_hex, shard_to_hex,
    reconstruct_key, hash_key
)
from .audit_trail import AuditTrail
from .task_engine import TaskEngine

logger = logging.getLogger(__name__)


class IssuingLayer:
    """
    Trusted component that:
    - Holds issuer_shard (in task JSON, never distributed to agents)
    - Manages step shards in vault filesystem (Phase 1)
    - Validates step results
    - Issues signed step tokens upon successful completion
    - Logs all events to audit trail
    - Optionally authenticates agents (Phase 5)
    """

    def __init__(self, task_engine: TaskEngine):
        self.task_engine = task_engine

    # ------------------------------------------------------------------
    # Phase 5: Agent authentication helper
    # ------------------------------------------------------------------
    def verify_agent_auth(self, agent_id: str, payload: str,
                          signature: str) -> bool:
        """Verify an agent's HMAC-signed request before token issuance.

        Returns True if the agent is registered and the signature is valid.
        """
        return self.task_engine.authenticate_agent_request(
            agent_id, payload, signature
        )

    # ------------------------------------------------------------------
    # Token issuance
    # ------------------------------------------------------------------
    def issue_step_token(self, task_id: str, step_number: int,
                         result: Any, result_hash: str,
                         agent_id: str = None,
                         agent_signature: str = None) -> Optional[Dict[str, Any]]:
        """
        Validate step result and issue signed token.

        Phase 1: Step shard read from vault, NOT from task JSON.
        Phase 2: Accepts steps in 'pending' or 'running' state.
        Phase 5: If agent_id is provided, verifies auth before issuance.
        """
        # Phase 5: optional agent authentication
        if agent_id is not None:
            payload = f"{task_id}:{step_number}:{result_hash}"
            if agent_signature is None:
                raise ValueError("agent_signature required when agent_id is provided")
            if not self.verify_agent_auth(agent_id, payload, agent_signature):
                raise PermissionError(
                    f"Agent '{agent_id}' authentication failed for step {step_number}"
                )
        else:
            logger.debug(
                "issue_step_token called without agent_id for task %s step %d "
                "(auth not enforced)", task_id, step_number
            )

        task = self.task_engine.get_task(task_id)
        step = task["steps"][step_number - 1]

        # Phase 2: allow both 'pending' and 'running' states
        if step["state"] not in ("pending", "running"):
            raise ValueError(
                f"Step {step_number} is not pending/running (current: {step['state']})"
            )

        # Generate token ID
        token_id = f"tok_{uuid.uuid4().hex[:12]}"

        # Phase 1: Load shards from vault for signing
        issuer_shard = shard_from_hex(task["issuer_shard"])
        step_shard = self.task_engine.get_step_shard(task_id, step_number)

        # Create token payload
        token = {
            "task_id": task_id,
            "step_number": step_number,
            "step_name": step["name"],
            "shard_hash": hash_key(step_shard),  # Phase 1: only expose hash
            "result_hash": result_hash,
            "token_id": token_id,
            "issued_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        }

        # Sign token with combined key material
        signing_key = issuer_shard + step_shard
        token["issuer_signature"] = hmac_sign(
            signing_key,
            f"{task_id}:{step_number}:{step['name']}:{result_hash}",
        )

        # Update task state
        step["state"] = "complete"
        step["token_id"] = token_id
        step["result_hash"] = result_hash
        step["idempotency_key"] = None  # clear running key
        task["completed_steps"] += 1

        # Check if all steps complete
        if task["completed_steps"] == task["num_steps"]:
            task["state"] = "reconstructing"
        else:
            task["state"] = "in_progress"

        self.task_engine.save_task(task)

        # Log to audit (Phase 4: use audit key, not master key)
        master_key = self._reconstruct_master_key(task)
        audit_key = self.task_engine._derive_audit_key(master_key)
        audit = AuditTrail(
            self.task_engine.data_dir / "audit" / f"{task_id}.jsonl", audit_key
        )
        audit.log(
            "STEP_COMPLETE",
            task_id,
            step_number=step_number,
            result_hash=result_hash,
            token_id=token_id,
        )

        return token

    def reconstruct_master_key(self, task_id: str) -> Optional[bytes]:
        """
        Reconstruct master key from all step tokens + issuer shard.
        Phase 1: Reads step shards from vault.
        Returns key bytes or None if incomplete.
        """
        task = self.task_engine.get_task(task_id)

        if task["completed_steps"] != task["num_steps"]:
            return None

        # Phase 1: Collect all shards from vault
        shards = [
            self.task_engine.get_step_shard(task_id, step["step_number"])
            for step in task["steps"]
        ]
        shards.append(shard_from_hex(task["issuer_shard"]))

        reconstructed = reconstruct_key(shards)

        # Verify hash
        expected_hash = hash_key(reconstructed)

        if expected_hash != task["key_hash"]:
            # Tamper detected!
            master_key = self._reconstruct_master_key(task)
            audit_key = self.task_engine._derive_audit_key(master_key)
            audit = AuditTrail(
                self.task_engine.data_dir / "audit" / f"{task_id}.jsonl",
                audit_key,
            )
            audit.log(
                "RECONSTRUCTION_FAILED",
                task_id,
                reason="hash_mismatch",
                expected=task["key_hash"],
                got=expected_hash,
            )
            task["state"] = "reconstruction_failed"
            self.task_engine.save_task(task)
            return None

        # Success
        task["state"] = "task_complete"
        self.task_engine.save_task(task)

        master_key = self._reconstruct_master_key(task)
        audit_key = self.task_engine._derive_audit_key(master_key)
        audit = AuditTrail(
            self.task_engine.data_dir / "audit" / f"{task_id}.jsonl", audit_key
        )
        audit.log("TASK_COMPLETE", task_id, key_hash=task["key_hash"])

        return reconstructed

    def _reconstruct_master_key(self, task) -> bytes:
        """Internal: reconstruct key from task data for audit signing.
        Phase 1: Reads step shards from vault."""
        shards = [
            self.task_engine.get_step_shard(task["task_id"], step["step_number"])
            for step in task["steps"]
        ]
        shards.append(shard_from_hex(task["issuer_shard"]))
        return reconstruct_key(shards)

    def fail_step(self, task_id: str, step_number: int, error: str):
        """Mark step as failed."""
        task = self.task_engine.get_task(task_id)
        step = task["steps"][step_number - 1]

        step["state"] = "failed"
        step["error"] = error
        step["retry_count"] += 1
        step["idempotency_key"] = None  # clear running key
        task["failed_steps"] += 1
        task["state"] = "paused"

        self.task_engine.save_task(task)

        # Log with audit key
        master_key = self._reconstruct_master_key(task)
        audit_key = self.task_engine._derive_audit_key(master_key)
        audit = AuditTrail(
            self.task_engine.data_dir / "audit" / f"{task_id}.jsonl", audit_key
        )
        audit.log("STEP_FAILED", task_id, step_number=step_number, error=error)

FILE:agentpathfinder/pathfinder_core.py
"""AgentPathfinder v2 — Core cryptography and shard management."""
import os
import hmac
import hashlib
import secrets
from typing import List, Tuple


def generate_master_key() -> bytes:
    """Generate 256-bit master key."""
    return secrets.token_bytes(32)


def split_key(master_key: bytes, num_steps: int) -> Tuple[List[bytes], bytes]:
    """
    Split master_key into (num_steps + 1) shards via XOR.
    
    Returns:
        (step_shards, issuer_shard) where:
        - step_shards: List[num_steps] — one per step
        - issuer_shard: bytes — final fragment held by issuer
        
    Reconstruction: XOR all shards together = master_key
    """
    if len(master_key) != 32:
        raise ValueError("Master key must be 32 bytes (256 bits)")
    
    total_shards = num_steps + 1
    
    # Generate N random shards for steps
    step_shards = [secrets.token_bytes(32) for _ in range(num_steps)]
    
    # Compute issuer shard: K XOR s1 XOR s2 XOR ... XOR sN
    issuer_shard = master_key
    for shard in step_shards:
        issuer_shard = bytes(a ^ b for a, b in zip(issuer_shard, shard))
    
    return step_shards, issuer_shard


def reconstruct_key(shards: List[bytes]) -> bytes:
    """Reconstruct master key from all shards via XOR."""
    if not shards:
        raise ValueError("No shards provided")
    
    key = shards[0]
    for shard in shards[1:]:
        key = bytes(a ^ b for a, b in zip(key, shard))
    
    return key


def hmac_sign(key: bytes, message: str) -> str:
    """HMAC-SHA256 sign a message. Returns hex string."""
    return hmac.new(key, message.encode(), hashlib.sha256).hexdigest()


def verify_hmac(key: bytes, message: str, signature: str) -> bool:
    """Verify HMAC-SHA256 signature."""
    expected = hmac_sign(key, message)
    return hmac.compare_digest(expected, signature)


def hash_key(key: bytes) -> str:
    """SHA-256 hash of key for public reference (never expose raw key)."""
    return hashlib.sha256(key).hexdigest()


def shard_to_hex(shard: bytes) -> str:
    return shard.hex()


def shard_from_hex(hex_str: str) -> bytes:
    return bytes.fromhex(hex_str)


def derive_key(master_key: bytes, context: bytes) -> bytes:
    """Derive a sub-key from master_key using HMAC-SHA256.

    This is a simple single-purpose KDF: HMAC-SHA256(master_key, context).
    Used to produce audit signing keys, agent API keys, etc. without
    exposing the master key itself.
    """
    return hmac.new(master_key, context, hashlib.sha256).digest()

FILE:agentpathfinder/task_engine.py
"""AgentPathfinder v2 — Task decomposition and state management (Phases 1-5)."""
import json
import os
import tempfile
import uuid
import time
import secrets
import logging
from contextlib import contextmanager
from pathlib import Path
from typing import Dict, List, Any, Optional
from enum import Enum

try:
    import fcntl
    _HAS_FCNTL = True
except ImportError:
    _HAS_FCNTL = False  # Windows — advisory locking disabled

from .pathfinder_core import (
    generate_master_key, split_key, hash_key,
    shard_to_hex, shard_from_hex, reconstruct_key,
    hmac_sign, derive_key
)
from .audit_trail import AuditTrail

logger = logging.getLogger(__name__)


class TaskState(Enum):
    REGISTERED = "registered"
    DISPATCHED = "dispatched"
    IN_PROGRESS = "in_progress"
    STEP_COMPLETE = "step_complete"
    STEP_FAILED = "step_failed"
    STEP_RUNNING = "step_running"       # Phase 2: a step is actively executing
    PAUSED = "paused"
    RECONSTRUCTING = "reconstructing"
    TASK_COMPLETE = "task_complete"
    RECONSTRUCTION_FAILED = "reconstruction_failed"
    ABORTED = "aborted"


class TaskEngine:
    """Manages task lifecycle: creation, execution, state transitions."""

    def __init__(self, data_dir: Optional[Path] = None):
        if data_dir is None:
            data_dir = Path.home() / ".agentpathfinder" / "pathfinder_data"
        self.data_dir = Path(data_dir)
        self.tasks_dir = self.data_dir / "tasks"
        self.vault_dir = self.data_dir / "vault"
        self.agents_dir = self.data_dir / "agents"
        self.tasks_dir.mkdir(parents=True, exist_ok=True)
        self.vault_dir.mkdir(parents=True, exist_ok=True)
        self.agents_dir.mkdir(parents=True, exist_ok=True)

    # ------------------------------------------------------------------
    # Atomic writes (Phase 2)
    # ------------------------------------------------------------------
    def _atomic_write(self, path, data):
        """Atomic file write: write to temp + os.rename (same filesystem)."""
        path = Path(path)
        path.parent.mkdir(parents=True, exist_ok=True)
        fd, tmp_path = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp")
        try:
            if isinstance(data, str):
                os.write(fd, data.encode())
            else:
                os.write(fd, data)
            os.fsync(fd)
            os.close(fd)
            os.rename(tmp_path, str(path))
        except BaseException:
            try:
                os.close(fd)
            except OSError:
                pass
            try:
                os.unlink(tmp_path)
            except OSError:
                pass
            raise

    # ------------------------------------------------------------------
    # File locking (Phase 3)
    # ------------------------------------------------------------------
    @contextmanager
    def _lock_task(self, task_id: str):
        """Exclusive advisory lock on a task's lock file (no-op on Windows)."""
        if not _HAS_FCNTL:
            yield  # Windows — no advisory locking
            return
        lock_path = self.tasks_dir / f"{task_id}.lock"
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        lock_fd = open(lock_path, "w")
        try:
            fcntl.flock(lock_fd, fcntl.LOCK_EX)
            yield
        finally:
            fcntl.flock(lock_fd, fcntl.LOCK_UN)
            lock_fd.close()

    @contextmanager
    def _lock_vault_shard(self, task_id: str, step_number: int):
        """Exclusive advisory lock on a vault shard file (no-op on Windows)."""
        if not _HAS_FCNTL:
            yield  # Windows — no advisory locking
            return
        lock_dir = self.vault_dir / task_id
        lock_dir.mkdir(parents=True, exist_ok=True)
        lock_path = lock_dir / f"{step_number}.shard.lock"
        lock_fd = open(lock_path, "w")
        try:
            fcntl.flock(lock_fd, fcntl.LOCK_EX)
            yield
        finally:
            fcntl.flock(lock_fd, fcntl.LOCK_UN)
            lock_fd.close()

    # ------------------------------------------------------------------
    # Vault helpers
    # ------------------------------------------------------------------
    def _get_vault_path(self, task_id: str, step_number: int) -> Path:
        """Get the filesystem path for a step's shard vault."""
        task_vault = self.vault_dir / task_id
        task_vault.mkdir(parents=True, exist_ok=True)
        return task_vault / f"{step_number}.shard"

    def _write_shard_to_vault(self, task_id: str, step_number: int, shard: bytes):
        """Write a step shard to the vault filesystem (atomic + locked)."""
        with self._lock_vault_shard(task_id, step_number):
            vault_path = self._get_vault_path(task_id, step_number)
            self._atomic_write(vault_path, shard)

    def _read_shard_from_vault(self, task_id: str, step_number: int) -> bytes:
        """Read a step shard from the vault filesystem (locked)."""
        with self._lock_vault_shard(task_id, step_number):
            vault_path = self._get_vault_path(task_id, step_number)
            if not vault_path.exists():
                raise FileNotFoundError(
                    f"Shard not found in vault for task {task_id} step {step_number}"
                )
            return vault_path.read_bytes()

    # ------------------------------------------------------------------
    # Audit key derivation (Phase 4)
    # ------------------------------------------------------------------
    def _derive_audit_key(self, master_key: bytes) -> bytes:
        """Derive a separate audit signing key from the master key.

        Phase 4: The audit trail must never have access to the master key.
        We use derive_key(master_key, b'audit_signing_key').
        """
        return derive_key(master_key, b"audit_signing_key")

    def _reconstruct_master_key(self, task: Dict[str, Any]) -> bytes:
        """Reconstruct master key from vault shards + issuer shard."""
        shards = [
            self._read_shard_from_vault(task["task_id"], step["step_number"])
            for step in task["steps"]
        ]
        shards.append(shard_from_hex(task["issuer_shard"]))
        return reconstruct_key(shards)

    # ------------------------------------------------------------------
    # Core CRUD
    # ------------------------------------------------------------------
    def create_task(self, name: str, steps: List[Dict[str, Any]]) -> str:
        """Create a new task. Step shards written to vault, NOT in task JSON."""
        task_id = str(uuid.uuid4())
        num_steps = len(steps)

        # Generate master key and shards
        master_key = generate_master_key()
        step_shards, issuer_shard = split_key(master_key, num_steps)

        # Phase 1: Write step shards to vault (atomic, Phase 2)
        for i, shard in enumerate(step_shards, start=1):
            self._write_shard_to_vault(task_id, i, shard)

        # Build task metadata (NO step shards in task JSON)
        task = {
            "task_id": task_id,
            "name": name,
            "state": TaskState.REGISTERED.value,
            "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "num_steps": num_steps,
            "key_hash": hash_key(master_key),
            "issuer_shard": shard_to_hex(issuer_shard),
            "completed_steps": 0,
            "failed_steps": 0,
            "steps": [
                {
                    "step_number": i + 1,
                    "name": step["name"],
                    "state": "pending",
                    "token_id": None,
                    "result_hash": None,
                    "error": None,
                    "retry_count": 0,
                    "idempotency_key": None,    # Phase 2
                }
                for i, step in enumerate(steps)
            ],
        }

        # Save task metadata (atomic, Phase 2)
        task_path = self.tasks_dir / f"{task_id}.json"
        self._atomic_write(task_path, json.dumps(task, indent=2, default=str))

        # Phase 4: Derive audit key from master key
        audit_key = self._derive_audit_key(master_key)

        # Initialize audit trail
        audit = AuditTrail(
            self.data_dir / "audit" / f"{task_id}.jsonl", audit_key
        )
        audit.log(
            "TASK_REGISTERED",
            task_id,
            name=name,
            steps=num_steps,
            key_hash=task["key_hash"],
        )

        return task_id

    def get_task(self, task_id: str) -> Dict[str, Any]:
        """Load task metadata by ID."""
        task_path = self.tasks_dir / f"{task_id}.json"
        if not task_path.exists():
            raise ValueError(f"Task {task_id} not found")
        with open(task_path) as f:
            return json.load(f)

    def _save_task_unlocked(self, task: Dict[str, Any]):
        """Save task state atomically (no lock — caller must hold lock)."""
        task_path = self.tasks_dir / f"{task['task_id']}.json"
        self._atomic_write(task_path, json.dumps(task, indent=2, default=str))

    def save_task(self, task: Dict[str, Any]):
        """Save task state (atomic + locked, Phases 2-3)."""
        with self._lock_task(task["task_id"]):
            self._save_task_unlocked(task)

    def get_step_shard(self, task_id: str, step_number: int) -> bytes:
        """Retrieve a step shard from the vault."""
        return self._read_shard_from_vault(task_id, step_number)

    # ------------------------------------------------------------------
    # Step state transitions (Phase 2)
    # ------------------------------------------------------------------
    def set_step_running(self, task_id: str, step_number: int,
                         idempotency_key: str) -> str:
        """Mark step as 'running' with an idempotency key.

        Returns the idempotency_key on success.
        Raises ValueError if the step is already running with a *different* key
        (concurrent duplicate) or if the step is not in a launchable state.
        """
        with self._lock_task(task_id):
            task = self.get_task(task_id)
            step = task["steps"][step_number - 1]

            if step["state"] == "running":
                if step.get("idempotency_key") == idempotency_key:
                    # Same execution — idempotent, skip
                    return idempotency_key
                raise ValueError(
                    f"Step {step_number} already running with different "
                    f"idempotency key (conflict)"
                )

            if step["state"] not in ("pending",):
                raise ValueError(
                    f"Step {step_number} cannot transition to running "
                    f"(current: {step['state']})"
                )

            step["state"] = "running"
            step["idempotency_key"] = idempotency_key
            task["state"] = TaskState.IN_PROGRESS.value

            self._save_task_unlocked(task)

        return idempotency_key

    def detect_crashed_steps(self, task_id: str) -> List[Dict[str, Any]]:
        """Find steps stuck in 'running' state (crash recovery).

        Returns a list of step dicts that are still in 'running' state.
        The caller should decide whether to reset them to 'pending' or
        mark them as 'failed'.
        """
        task = self.get_task(task_id)
        return [
            step for step in task["steps"]
            if step["state"] == "running"
        ]

    def reset_running_step(self, task_id: str, step_number: int):
        """Reset a stuck 'running' step back to 'pending' (crash recovery)."""
        with self._lock_task(task_id):
            task = self.get_task(task_id)
            step = task["steps"][step_number - 1]
            if step["state"] != "running":
                raise ValueError(
                    f"Step {step_number} is not running (current: {step['state']})"
                )
            step["state"] = "pending"
            step["idempotency_key"] = None
            self._save_task_unlocked(task)

    # ------------------------------------------------------------------
    # Status & recovery
    # ------------------------------------------------------------------
    def get_status(self, task_id: str) -> Dict[str, Any]:
        """Get human-readable task status."""
        task = self.get_task(task_id)

        step_statuses = []
        for step in task["steps"]:
            step_statuses.append(
                {
                    "step_number": step["step_number"],
                    "name": step["name"],
                    "state": step["state"],
                    "token_id": step.get("token_id"),
                    "error": step.get("error"),
                }
            )

        return {
            "task_id": task_id,
            "name": task["name"],
            "overall_state": task["state"],
            "progress": f"{task['completed_steps']}/{task['num_steps']}",
            "steps": step_statuses,
            "all_complete": task["completed_steps"] == task["num_steps"],
        }

    def reset_step(self, task_id: str, step_number: int) -> Dict[str, Any]:
        """Reset a failed step to pending so it can be retried."""
        task = self.get_task(task_id)
        step = task["steps"][step_number - 1]

        if step["state"] != "failed":
            raise ValueError(
                f"Step {step_number} is not failed (current: {step['state']})"
            )

        step["state"] = "pending"
        step["error"] = None
        step["retry_count"] = 0
        step["idempotency_key"] = None

        task["failed_steps"] = max(0, task["failed_steps"] - 1)
        task["state"] = "in_progress"

        self.save_task(task)

        # Log retry attempt with audit key
        master_key = self._reconstruct_master_key(task)
        audit_key = self._derive_audit_key(master_key)
        audit = AuditTrail(
            self.data_dir / "audit" / f"{task_id}.jsonl", audit_key
        )
        audit.log("STEP_RETRY_INITIATED", task_id, step_number=step_number)

        return step

    def resume_from_failure(self, task_id: str) -> Optional[int]:
        """Find the first failed step and return its number for retry."""
        task = self.get_task(task_id)

        for step in task["steps"]:
            if step["state"] == "failed":
                self.reset_step(task_id, step["step_number"])
                return step["step_number"]

        return None

    # ------------------------------------------------------------------
    # Agent registry (Phase 5)
    # ------------------------------------------------------------------
    def _agents_file(self) -> Path:
        return self.agents_dir / "registry.json"

    def _load_agents(self) -> Dict[str, str]:
        """Load agent registry {agent_id: api_key_hex}."""
        path = self._agents_file()
        if not path.exists():
            return {}
        with open(path) as f:
            return json.load(f)

    def _save_agents(self, agents: Dict[str, str]):
        self._atomic_write(self._agents_file(), json.dumps(agents, indent=2))

    def register_agent(self, agent_id: str) -> str:
        """Register an agent and return its API key (hex-encoded shared secret).

        If the agent already exists, the existing key is returned.
        """
        agents = self._load_agents()
        if agent_id in agents:
            return agents[agent_id]

        api_key = secrets.token_hex(32)  # 256-bit shared secret
        agents[agent_id] = api_key
        self._save_agents(agents)
        logger.info("Registered agent %s", agent_id)
        return api_key

    def verify_agent(self, agent_id: str, api_key: str) -> bool:
        """Verify agent credentials (constant-time comparison)."""
        agents = self._load_agents()
        expected = agents.get(agent_id)
        if expected is None:
            return False
        import hmac as _hmac
        return _hmac.compare_digest(expected, api_key)

    def authenticate_agent_request(self, agent_id: str, payload: str,
                                   signature: str) -> bool:
        """Verify an HMAC-signed request from an agent.

        The agent signs the payload with its API key (as raw bytes).
        """
        agents = self._load_agents()
        api_key_hex = agents.get(agent_id)
        if api_key_hex is None:
            return False
        api_key_bytes = bytes.fromhex(api_key_hex)
        return hmac_sign(api_key_bytes, payload) == signature  # constant-time inside hmac_sign/verify

FILE:requirements.txt
# Core: zero dependencies (stdlib only)

# Dashboard (optional)
flask>=2.0.0

FILE:scripts/__init__.py

FILE:scripts/dashboard_static.py
"""AgentPathfinder dashboard static HTML generator.

Generates dashboard.html showing task status, audit trails, and metrics.
"""
import json
import subprocess
from pathlib import Path
from typing import Dict, Any, List

# ── paths ──
DASHBOARD_DIR = Path(__file__).parent
TASK_DIR = Path.home() / ".agentpathfinder" / "pathfinder_data" / "tasks"
AUDIT_DIR = Path.home() / ".agentpathfinder" / "pathfinder_data" / "audit"
METRICS_FILE = Path.home() / ".agentpathfinder" / "metrics.json"

HTML_TEMPLATE = """<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>AgentPathfinder Dashboard</title>
<style>
:root {{
  --navy: #0F1724; --navy-light: #1E293B; --blue: #2563EB;
  --text: #E2E8F0; --text-soft: #94A3B8; --text-dim: #64748B;
  --success: #34D399; --error: #F87171; --warning: #FBBF24;
  --card-bg: #0B1120; --border: rgba(255,255,255,0.06);
}}
* {{ margin:0; padding:0; box-sizing:border-box; }}
body {{
  font-family: -apple-system, BlinkMacSystemFont, 'SF Pro', 'Segoe UI', sans-serif;
  background: var(--navy); color: var(--text); min-height: 100vh;
}}
.container {{ max-width: 1400px; margin: 0 auto; padding: 0 24px; }}
.header {{
  background: var(--card-bg); border-bottom: 1px solid var(--border);
  padding: 16px 0; position: sticky; top: 0; z-index: 100;
}}
.header-inner {{ display: flex; justify-content: space-between; align-items: center; }}
.brand-icon {{
  width: 32px; height: 32px; background: linear-gradient(135deg, var(--blue), #3B82F6);
  border-radius: 8px; display: grid; place-items: center; font-size: 16px;
}}
.hero {{ padding: 40px 0 24px; }}
.hero-grid {{ display: grid; grid-template-columns: repeat(4, 1fr); gap: 16px; }}
.stat-card {{
  background: var(--card-bg); border: 1px solid var(--border);
  border-radius: 12px; padding: 20px;
}}
.stat-card::before {{
  content: ""; position: absolute; top: 0; left: 0; right: 0; height: 2px;
  background: linear-gradient(90deg, var(--blue), transparent 60%);
}}
.stat-label {{ font-size: 12px; color: var(--text-dim); text-transform: uppercase; letter-spacing: 0.5px; margin-bottom: 8px; }}
.stat-value {{ font-size: 32px; font-weight: 700; letter-spacing: -1px; }}
.content {{ padding: 24px 0 60px; }}
.grid-2 {{ display: grid; grid-template-columns: 1fr 1fr; gap: 16px; }}
.card {{ background: var(--card-bg); border: 1px solid var(--border); border-radius: 12px; padding: 20px; }}
.footer {{ border-top: 1px solid var(--border); padding: 20px 0; text-align: center; color: var(--text-dim); font-size: 12px; }}
</style>
</head>
<body>
<div class="header">
  <div class="container">
    <div class="header-inner">
      <div style="display:flex;align-items:center;gap:10px">
        <div class="brand-icon">🎯</div>
        <div style="font-size:18px;font-weight:700">AgentPathfinder</div>
      </div>
    </div>
  </div>
</div>

<div class="container">
  <div class="hero">
    <div class="hero-grid">
      <div class="stat-card">
        <div class="stat-label">Active Tasks</div>
        <div class="stat-value">{tasks_count}</div>
      </div>
      <div class="stat-card">
        <div class="stat-label">Audit Events</div>
        <div class="stat-value">{audit_count}</div>
      </div>
      <div class="stat-card">
        <div class="stat-label">Agents</div>
        <div class="stat-value">{agent_count}</div>
      </div>
    </div>
  </div>

  <div class="content">
    <div class="grid-2">
      <div class="card">
        <h3>Tasks</h3>
        {tasks_html}
      </div>
      <div class="card">
        <h3>Audit Trail</h3>
        {audit_html}
      </div>
    </div>
  </div>
</div>

<div class="footer">
  <div class="container">
    AgentPathfinder — Tamper-evident task tracking<br>
    <span style="color:var(--text-dim)">Version 1.2.6 • MIT License</span>
  </div>
</div>
</body>
</html>"""


def _count_json(dir_path: Path) -> int:
    if not dir_path.exists():
        return 0
    return len(list(dir_path.glob("*.json")))


def _count_jsonl_lines(dir_path: Path) -> int:
    count = 0
    for f in dir_path.glob("*.jsonl"):
        count += sum(1 for line in open(f) if line.strip())
    return count


def _load_tasks_html() -> str:
    if not TASK_DIR.exists() or not list(TASK_DIR.glob("*.json")):
        return '<div style="color:var(--text-dim)">No tasks yet.</div>'
    
    tasks = []
    for f in sorted(TASK_DIR.glob("*.json")):
        data = json.loads(f.read_text())
        tasks.append(data)
    
    html = '<ul style="list-style:none;padding:0">'
    for t in tasks[:10]:
        html += f'''<li style="padding:8px 0;border-bottom:1px solid var(--border)">
            <strong>{t.get("name", "Untitled")}</strong>
            <span style="color:var(--text-dim);font-size:12px"> — {t.get("status", "unknown")}</span>
        </li>'''
    html += '</ul>'
    return html


def _load_audit_html() -> str:
    if not AUDIT_DIR.exists():
        return '<div style="color:var(--text-dim)">No audit events yet.</div>'
    
    events = []
    for f in sorted(AUDIT_DIR.glob("*.jsonl")):
        for line in open(f):
            line = line.strip()
            if not line:
                continue
            events.append(json.loads(line))
    
    if not events:
        return '<div style="color:var(--text-dim)">No audit events yet.</div>'
    
    html = '<ul style="list-style:none;padding:0">'
    for e in events[-10:]:
        html += f'''<li style="padding:8px 0;border-bottom:1px solid var(--border)">
            <span style="font-size:12px;color:var(--text-dim)">{e.get("timestamp", "?")}</span>
            <div>{e.get("event", "unknown")} — task: {e.get("task_id", "?")[:12]}</div>
        </li>'''
    html += '</ul>'
    return html


def generate_dashboard(output_path: str = "report.html") -> Path:
    out = DASHBOARD_DIR / output_path
    
    tasks_count = _count_json(TASK_DIR)
    audit_count = _count_jsonl_lines(AUDIT_DIR)
    agent_count = _count_json(Path.home() / ".agentpathfinder" / "pathfinder_data" / "agents")
    
    html = HTML_TEMPLATE.format(
        tasks_count=tasks_count,
        audit_count=audit_count,
        agent_count=agent_count,
        tasks_html=_load_tasks_html(),
        audit_html=_load_audit_html(),
    )
    
    out.write_text(html)
    return out


if __name__ == "__main__":
    path = generate_dashboard()
    print(f"Dashboard written to: {path}")

FILE:scripts/pathfinder_client.py
#!/usr/bin/env python3
"""AgentPathfinder v2 — ClawHub CLI client and SDK wrapper.

Deterministic task orchestration with cryptographic sharding.
Usage:
    from pathfinder_client import PathfinderClient
    client = PathfinderClient()
    task_id = client.create("deploy", ["build", "push", "verify"])
    client.run(task_id)
    print(client.status(task_id))
"""

import argparse
import json
import os
import sys
import time
import subprocess
from pathlib import Path
from typing import List, Dict, Any, Optional

# ── Import resolution ───────────────────────────────────────────────
_SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(_SCRIPT_DIR.parent))
from agentpathfinder import (
    TaskEngine, IssuingLayer, AgentRuntime, AuditTrail,
    generate_master_key, split_key, reconstruct_key,
    hmac_sign, verify_hmac, hash_key, derive_key,
)

# Import visual helpers
sys.path.insert(0, str(_SCRIPT_DIR))
from visual import (
    fmt_status, fmt_audit_event, fmt_step_complete, fmt_step_failed,
    fmt_task_complete, fmt_task_failed, fmt_reconstruct_ok, fmt_reconstruct_fail,
    fmt_agent_registered, fmt_dashboard_url, fmt_install_ready,
    fmt_crash_recovery, PASS, FAIL, SPINNER, INFO, badge_ok, badge_fail,
    green, red, yellow, bold, dim,
)

# ── Config ─────────────────────────────────────────────────────────
DATA_DIR = Path(os.getenv("PATHFINDER_DATA_DIR", "./pathfinder_data"))
DASHBOARD_PORT = int(os.getenv("PATHFINDER_DASHBOARD_PORT", "8080"))


class PathfinderClient:
    """SDK wrapper around AgentPathfinder v2 core."""

    def __init__(self, data_dir: Path = None):
        self.data_dir = Path(data_dir) if data_dir else DATA_DIR
        self.engine = TaskEngine(data_dir=self.data_dir)
        self.issuing = IssuingLayer(self.engine)
        self.runtime = AgentRuntime(self.engine, self.issuing)

    # ── Tasks ─────────────────────────────────────────────────────

    def create(self, name: str, steps: List[str]) -> str:
        """Create a new task. Returns the task_id."""
        step_specs = [{"name": s} for s in steps]
        return self.engine.create_task(name, step_specs)

    def run(self, task_id: str) -> Dict[str, Any]:
        """Simulate running all steps and attempt reconstruction."""
        task = self.engine.get_task(task_id)
        for step in task["steps"]:
            sn = step["step_number"]
            if step["state"] in ("pending", "running"):
                result_str = f"simulated_result_for_step_{sn}"
                result_hash = hash_key(result_str.encode())[:16]
                self.issuing.issue_step_token(task_id, sn, result_str, result_hash)
        self.issuing.reconstruct_master_key(task_id)
        return self.engine.get_status(task_id)

    def status(self, task_id: str) -> Dict[str, Any]:
        return self.engine.get_status(task_id)

    def audit(self, task_id: str) -> List[Dict[str, Any]]:
        task = self.engine.get_task(task_id)
        master_key = self.engine._reconstruct_master_key(task)
        audit_key = self.engine._derive_audit_key(master_key)
        audit = AuditTrail(
            self.engine.data_dir / "audit" / f"{task_id}.jsonl",
            audit_key,
        )
        return audit.read_trail(task_id)

    def reconstruct(self, task_id: str) -> Optional[bytes]:
        return self.issuing.reconstruct_master_key(task_id)

    # ── Agents ────────────────────────────────────────────────────

    def register_agent(self, agent_id: str) -> str:
        return self.engine.register_agent(agent_id)

    def verify_agent(self, agent_id: str, api_key: str) -> bool:
        return self.engine.verify_agent(agent_id, api_key)

    # ── Dashboard ────────────────────────────────────────────────

    def dashboard(self) -> None:
        """Generate static dashboard HTML."""
        dashboard_script = Path(__file__).parent / "dashboard_static.py"
        if not dashboard_script.exists():
            raise FileNotFoundError(f"Dashboard script not found: {dashboard_script}")
        subprocess.run([sys.executable, str(dashboard_script)], check=True)


# ── CLI Handlers with Visual Confirmations ────────────────────────

def cli_create(args):
    client = PathfinderClient()
    name = args.name
    steps = args.steps or ["step_1"]
    task_id = client.create(name, steps)
    print(f"{PASS} Task created: {bold(task_id)}")
    print(f"   {INFO} Name: {name}")
    print(f"   {INFO} Steps: {len(steps)}")
    return task_id


def cli_run(args):
    client = PathfinderClient()
    print(f"{WARN} SIMULATION MODE — No real code executed.")
    print(f"    Use Python SDK for production execution.")
    print(f"    Docs: github.com/CertainLogicAI/agentpathfinder#sdk")
    print(f"{SPINNER} Running task {dim(args.task_id)}...")
    status = client.run(args.task_id)
    print("")
    if status["overall_state"] == "task_complete":
        print(fmt_task_complete(status["name"], args.task_id))
        print(f"   {bold('Progress:')} {green(status['progress'])}")
        for step in status["steps"]:
            if step["state"] == "complete":
                print(f"   {fmt_step_complete(step['step_number'], step['name'], step.get('token_id',''))}")
    else:
        print(fmt_task_failed(status["name"], args.task_id, "Not all steps completed"))
        for step in status["steps"]:
            if step["state"] == "failed":
                print(f"   {fmt_step_failed(step['step_number'], step['name'], step.get('error','Unknown error'))}")
    print("")
    print(fmt_status(status))
    return status


def cli_status(args):
    client = PathfinderClient()
    try:
        status = client.status(args.task_id)
        print(fmt_status(status))
        # Crash recovery check
        if any(s["state"] == "running" for s in status["steps"]):
            print("")
            print(f"{FAIL} {bold('WARNING:')} Steps stuck in 'running' — possible crash.")
            print(f"   {INFO} Use {bold('pf reset-step')} or check logs.")
    except ValueError as e:
        print(f"{FAIL} {e}")


def cli_audit(args):
    client = PathfinderClient()
    try:
        events = client.audit(args.task_id)
        print(f"\n{bold('Audit Trail for')} {dim(args.task_id)}:")
        for ev in events:
            print(fmt_audit_event(ev))
        tampered = sum(1 for e in events if not e.get("tamper_ok", True))
        print("")
        if tampered == 0:
            print(badge_ok(f"All {len(events)} events verified"))
        else:
            print(badge_fail(f"{tampered} of {len(events)} events TAMPERED"))
    except Exception as e:
        print(f"{FAIL} Audit failed: {e}")


def cli_reconstruct(args):
    client = PathfinderClient()
    key = client.reconstruct(args.task_id)
    if key:
        task = client.engine.get_task(args.task_id)
        print(fmt_reconstruct_ok(task["key_hash"]))
    else:
        print(fmt_reconstruct_fail())
    return key


def cli_register_agent(args):
    client = PathfinderClient()
    api_key = client.register_agent(args.agent_id)
    print(fmt_agent_registered(args.agent_id, api_key))
    return api_key


def cli_dashboard(args):
    """Generate the static HTML dashboard."""
    dashboard_script = Path(__file__).parent / "dashboard_static.py"
    if not dashboard_script.exists():
        print(f"{FAIL} Dashboard generator not found: {dashboard_script}")
        sys.exit(1)
    print(f"{SPINNER} Generating dashboard...")
    try:
        result = subprocess.run(
            [sys.executable, str(dashboard_script)],
            capture_output=True, text=True, check=True
        )
        # Parse output for file path
        for line in result.stdout.split('\n'):
            if 'Dashboard written to:' in line:
                print(f"{PASS} {line}")
                break
        else:
            print(result.stdout)
    except subprocess.CalledProcessError as e:
        print(f"{FAIL} Dashboard generation failed: {e.stderr}")
    except Exception as e:
        print(f"{FAIL} Dashboard failed: {e}")


def cli_install(args):
    """One-command setup: init data dir, print ready message."""
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    (DATA_DIR / "tasks").mkdir(exist_ok=True)
    (DATA_DIR / "vault").mkdir(exist_ok=True)
    (DATA_DIR / "audit").mkdir(exist_ok=True)
    (DATA_DIR / "agents").mkdir(exist_ok=True)
    print(fmt_install_ready())


def main():
    parser = argparse.ArgumentParser(
        description="AgentPathfinder v2 — Deterministic task orchestration"
    )
    sub = parser.add_subparsers(dest="command")

    # create
    p_create = sub.add_parser("create", help="Create a new task")
    p_create.add_argument("name", help="Task name")
    p_create.add_argument("steps", nargs="*", help="Step names")
    p_create.set_defaults(func=cli_create)

    # run
    p_run = sub.add_parser("run", help="Simulate running all steps")
    p_run.add_argument("task_id", help="Task ID")
    p_run.set_defaults(func=cli_run)

    # status
    p_status = sub.add_parser("status", help="Show task status")
    p_status.add_argument("task_id", help="Task ID")
    p_status.set_defaults(func=cli_status)

    # audit
    p_audit = sub.add_parser("audit", help="Show audit trail")
    p_audit.add_argument("task_id", help="Task ID")
    p_audit.set_defaults(func=cli_audit)

    # reconstruct
    p_recon = sub.add_parser("reconstruct", help="Reconstruct master key")
    p_recon.add_argument("task_id", help="Task ID")
    p_recon.set_defaults(func=cli_reconstruct)

    # register-agent
    p_agent = sub.add_parser("register-agent", help="Register a new agent")
    p_agent.add_argument("agent_id", help="Agent identifier")
    p_agent.set_defaults(func=cli_register_agent)

    # dashboard
    p_dash = sub.add_parser("dashboard", help="Generate static HTML dashboard")
    p_dash.set_defaults(func=cli_dashboard)

    # install / setup
    p_install = sub.add_parser("install", help="One-command setup")
    p_install.set_defaults(func=cli_install)

    args = parser.parse_args()
    if hasattr(args, "func"):
        try:
            args.func(args)
        except ValueError as e:
            print(f"{FAIL} {e}")
            sys.exit(1)
        except FileNotFoundError as e:
            print(f"{FAIL} {e}")
            sys.exit(1)
    else:
        parser.print_help()


if __name__ == "__main__":
    main()

FILE:scripts/report.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>AgentPathfinder Dashboard</title>
<style>
:root {
  --navy: #0F1724; --navy-light: #1E293B; --blue: #2563EB;
  --text: #E2E8F0; --text-soft: #94A3B8; --text-dim: #64748B;
  --success: #34D399; --error: #F87171; --warning: #FBBF24;
  --card-bg: #0B1120; --border: rgba(255,255,255,0.06);
}
* { margin:0; padding:0; box-sizing:border-box; }
body {
  font-family: -apple-system, BlinkMacSystemFont, 'SF Pro', 'Segoe UI', sans-serif;
  background: var(--navy); color: var(--text); min-height: 100vh;
}
.container { max-width: 1400px; margin: 0 auto; padding: 0 24px; }
.header {
  background: var(--card-bg); border-bottom: 1px solid var(--border);
  padding: 16px 0; position: sticky; top: 0; z-index: 100;
}
.header-inner { display: flex; justify-content: space-between; align-items: center; }
.brand-icon {
  width: 32px; height: 32px; background: linear-gradient(135deg, var(--blue), #3B82F6);
  border-radius: 8px; display: grid; place-items: center; font-size: 16px;
}
.hero { padding: 40px 0 24px; }
.hero-grid { display: grid; grid-template-columns: repeat(4, 1fr); gap: 16px; }
.stat-card {
  background: var(--card-bg); border: 1px solid var(--border);
  border-radius: 12px; padding: 20px;
}
.stat-card::before {
  content: ""; position: absolute; top: 0; left: 0; right: 0; height: 2px;
  background: linear-gradient(90deg, var(--blue), transparent 60%);
}
.stat-label { font-size: 12px; color: var(--text-dim); text-transform: uppercase; letter-spacing: 0.5px; margin-bottom: 8px; }
.stat-value { font-size: 32px; font-weight: 700; letter-spacing: -1px; }
.content { padding: 24px 0 60px; }
.grid-2 { display: grid; grid-template-columns: 1fr 1fr; gap: 16px; }
.card { background: var(--card-bg); border: 1px solid var(--border); border-radius: 12px; padding: 20px; }
.footer { border-top: 1px solid var(--border); padding: 20px 0; text-align: center; color: var(--text-dim); font-size: 12px; }
</style>
</head>
<body>
<div class="header">
  <div class="container">
    <div class="header-inner">
      <div style="display:flex;align-items:center;gap:10px">
        <div class="brand-icon">🎯</div>
        <div style="font-size:18px;font-weight:700">AgentPathfinder</div>
      </div>
    </div>
  </div>
</div>

<div class="container">
  <div class="hero">
    <div class="hero-grid">
      <div class="stat-card">
        <div class="stat-label">Active Tasks</div>
        <div class="stat-value">0</div>
      </div>
      <div class="stat-card">
        <div class="stat-label">Audit Events</div>
        <div class="stat-value">0</div>
      </div>
      <div class="stat-card">
        <div class="stat-label">Agents</div>
        <div class="stat-value">0</div>
      </div>
    </div>
  </div>

  <div class="content">
    <div class="grid-2">
      <div class="card">
        <h3>Tasks</h3>
        <div style="color:var(--text-dim)">No tasks yet.</div>
      </div>
      <div class="card">
        <h3>Audit Trail</h3>
        <div style="color:var(--text-dim)">No audit events yet.</div>
      </div>
    </div>
  </div>
</div>

<div class="footer">
  <div class="container">
    AgentPathfinder — Tamper-evident task tracking<br>
    <span style="color:var(--text-dim)">Version 1.2.6 • MIT License</span>
  </div>
</div>
</body>
</html>
FILE:scripts/visual.py
"""AgentPathfinder v2 — Visual confirmation constants & formatting macros.

Every agent reply about task status uses these emoji/formatting helpers
for at-a-glance indicator: ✅ complete, ❌ failed, ⏳ running, etc.
"""

from typing import Dict, Any, List, Optional

# ── Core Status Emojis ──────────────────────────────────────────────
PASS         = "✅"
FAIL         = "❌"
WARN         = "⚠️"
SPINNER      = "⏳"
PENDING      = "○"
LOCK         = "🔒"
UNLOCK       = "🔓"
SHEILD       = "🛡️"
CHART        = "📊"
DASHBOARD    = "🖥️"
AGENT        = "🤖"
KEY          = "🔑"
CHECK        = "✓"
CROSS        = "✗"
ARROW        = "→"
BULLET       = "•"
STAR         = "⭐"
ROCKET       = "🚀"
INFO         = "ℹ️"
BELL         = "🔔"

# ── Step State Mapping ──────────────────────────────────────────────
STEP_ICONS = {
    "pending":   PENDING,
    "running":   SPINNER,
    "complete":  PASS,
    "failed":    FAIL,
}
STEP_COLORS = {
    "pending":   "#64748b",
    "running":   "#0ea5e9",
    "complete":  "#10b981",
    "failed":    "#ef4444",
}
STEP_EMOJI_COLOR = {
    "pending":   ("○", "dim"),
    "running":   ("⏳", "blue"),
    "complete":  ("✅", "green"),
    "failed":    ("❌", "red"),
}

# ── Task State Mapping ──────────────────────────────────────────────
TASK_ICONS = {
    "registered":            "📋",
    "in_progress":           SPINNER,
    "task_complete":         PASS,
    "paused":                WARN,
    "reconstruction_failed": FAIL,
    "aborted":               CROSS,
}

# ── Audit Result ───────────────────────────────────────────────────
AUDIT_OK   = f"{PASS} Audit integrity verified"
AUDIT_FAIL = f"{FAIL} Audit tampering detected"

# ── Reconstruction Result ──────────────────────────────────────────
RECON_OK   = f"{PASS} Key reconstructed successfully"
RECON_FAIL = f"{FAIL} Reconstruction failed — incomplete or tampered"

# ── Agent Auth ─────────────────────────────────────────────────────
AUTH_OK  = f"{PASS} Agent authenticated"
AUTH_FAIL = f"{FAIL} Agent authentication failed"

# ── Crash Recovery ─────────────────────────────────────────────────
CRASH_DETECTED = f"{WARN} Crash detected — step stuck in 'running'"
CRASH_RESET    = f"{PASS} Step reset to 'pending' for retry"

# ── Summary Badges ─────────────────────────────────────────────────
def badge_ok(text: str) -> str:
    return f"{PASS} {text}"

def badge_fail(text: str) -> str:
    return f"{FAIL} {text}"

def badge_warn(text: str):
    return f"{WARN} {text}"

def badge_info(text: str) -> str:
    return f"{INFO} {text}"

def badge_running(text: str) -> str:
    return f"{SPINNER} {text}"

# ── Terminal helpers (no colorama dep) ─────────────────────────────
ANSI = {
    "reset":    "\033[0m",
    "bold":     "\033[1m",
    "dim":      "\033[2m",
    "italic":   "\033[3m",
    "underline":"\033[4m",
    "red":      "\033[31m",
    "green":    "\033[32m",
    "yellow":   "\033[33m",
    "blue":     "\033[34m",
    "magenta":  "\033[35m",
    "cyan":     "\033[36m",
    "white":    "\033[37m",
    "bg_red":   "\033[41m",
    "bg_green": "\033[42m",
    "bg_yellow":"\033[43m",
}


def color(text: str, *codes: str) -> str:
    """Wrap text in ANSI color codes."""
    prefix = "".join(ANSI.get(c, "") for c in codes)
    return f"{prefix}{text}{ANSI['reset']}"


def dim(text: str) -> str:
    return color(text, "dim")


def bold(text: str) -> str:
    return color(text, "bold")


def green(text: str) -> str:
    return color(text, "green")


def red(text: str) -> str:
    return color(text, "red")


def yellow(text: str) -> str:
    return color(text, "yellow")


def blue(text: str) -> str:
    return color(text, "blue")


# ── Status Formatters ──────────────────────────────────────────────
def fmt_status(status: Dict[str, Any]) -> str:
    """Format a full task status dict into a visual, at-a-glance string."""
    lines = [
        "",
        f"{bold('Task:')} {status['name']} ({dim(status['task_id'])})",
        f"{bold('State:')} {fmt_state(status['overall_state'])}",
        f"{bold('Progress:')} {status['progress']}",
        "",
        bold("Steps:"),
    ]
    for step in status["steps"]:
        icon, color_name = STEP_EMOJI_COLOR.get(step["state"], ("?", "dim"))
        line = f"  {icon} Step {step['step_number']}: {step['name']}"
        if step.get("token_id"):
            line += f"  {dim('token:')}{dim(step['token_id'][:12])}…"
        if step.get("error"):
            line += f"  {red(step['error'][:60])}"
        lines.append(line)
    return "\n".join(lines)


def fmt_state(state: str) -> str:
    """Color-code a task state string."""
    mapping = {
        "task_complete":  green,
        "registered":     dim,
        "in_progress":    blue,
        "paused":         yellow,
        "reconstruction_failed": red,
        "aborted":        red,
    }
    return mapping.get(state, str)(state)


def fmt_audit_event(event: Dict[str, Any]) -> str:
    """Format a single audit event for display."""
    tamper = "OK" if event.get("tamper_ok") else "TAMPERED"
    ts = event.get("timestamp", "?")
    ev_type = event.get("event", "UNKNOWN")
    icon = PASS if event.get("tamper_ok") else FAIL
    return f"  [{ts}] {icon} {ev_type} [{tamper}]"


def fmt_step_complete(step_number: int, name: str, token_id: str = "") -> str:
    """Visual confirmation when a step completes."""
    tok = f" (token: {token_id[:12]}…)" if token_id else ""
    return f"{PASS} Step {step_number} complete: {name}{tok}"


def fmt_step_failed(step_number: int, name: str, error: str) -> str:
    """Visual indication when a step fails, with retry suggestion."""
    return (
        f"{FAIL} Step {step_number} failed: {name}\n"
        f"   {WARN} {error}\n"
        f"   {INFO} Suggestion: run  {bold('pf status <task_id>')}  then retry."
    )


def fmt_task_complete(name: str, task_id: str) -> str:
    return f"{PASS} {bold(name)} is complete! ID: {dim(task_id)}"


def fmt_task_failed(name: str, task_id: str, reason: str) -> str:
    return f"{FAIL} {bold(name)} failed. {reason}  ID: {dim(task_id)}"


def fmt_crash_recovery(step_number: int, name: str) -> str:
    return (
        f"{WARN} Crash recovery: Step {step_number} ({name}) was stuck in 'running'.\n"
        f"{CRASH_RESET}"
    )


def fmt_reconstruct_ok(key_hash: str) -> str:
    return f"{RECON_OK}\n   {dim('Key hash:')} {dim(key_hash[:16])}…"


def fmt_reconstruct_fail() -> str:
    return f"{RECON_FAIL}\n   {INFO} Not all steps finished. Check status."


def fmt_agent_registered(agent_id: str, api_key: str) -> str:
    masked = api_key[:8] + "…" + api_key[-8:]
    return (
        f"{PASS} Agent registered: {bold(agent_id)}\n"
        f"   {KEY} API key: {dim(masked)}"
    )


def fmt_dashboard_url(port: int) -> str:
    return f"{DASHBOARD} Dashboard live at {bold(f'http://localhost:{port}')}"


# ── Task Summary Visuals ─────────────────────────────────────────────
def fmt_task_summary(tasks: List[dict]) -> str:
    """Format task summary with visual badges."""
    count = len(tasks) if tasks else 0
    if not count:
        return f"  {INFO} No active tasks"
    
    complete = sum(1 for t in tasks if t.get("status") == "complete")
    lines = [
        "",
        f"{CHART} {bold('Task Summary')}",
        f"  {PASS} Active tasks:     {count}",
        f"  {PASS} Complete:         {complete}",
        f"  {SPINNER} In progress:      {count - complete}",
    ]
    return "\n".join(lines)


def fmt_install_ready() -> str:
    """The 'we're ready' banner."""
    return (
        f"\n{ROCKET} {bold(color('AgentPathfinder Ready!', 'green', 'bold'))}\n"
        f"   {PASS} Skill installed\n"
        f"   {PASS} Data directory initialized\n"
        f"   {DASHBOARD} Dashboard:  {bold('pf dashboard')}\n"
        f"   {INFO} Quick start:  {bold('pf create my_task step1 step2')}\n"
        f"\n{faint('All data stays in ~/.agentpathfinder — no external servers.')}\n"
    )


def fmt_box(title: str, body_lines: List[str], width: int = 58) -> str:
    """Draw a simple ASCII box."""
    top = "┌" + "─" * (width - 2) + "┐"
    mid = "│ " + title.ljust(width - 4) + " │"
    sep = "├" + "─" * (width - 2) + "┤"
    bot = "└" + "─" * (width - 2) + "┘"
    lines = [top, mid, sep]
    for line in body_lines:
        lines.append("│ " + line.ljust(width - 4) + " │")
    lines.append(bot)
    return "\n".join(lines)

FILE:setup.py
from setuptools import setup, find_packages

setup(
    name="agentpathfinder",
    version="1.0.1",
    description="Deterministic task orchestration with cryptographic sharding",
    author="CertainLogic",
    url="https://github.com/CertainLogicAI/agentpathfinder",
    packages=find_packages(),
    python_requires=">=3.10",
    entry_points={
        "console_scripts": [
            "pf=scripts.pathfinder_client:main",
        ],
    },
    install_requires=[],
    extras_require={
        "dashboard": ["flask>=2.0.0"],
    },
    license="MIT",
)

FILE:skill.json
{
  "schemaVersion": "2024-v2",
  "id": "certainlogicai.agentpathfinder",
  "displayName": "AgentPathfinder",
  "version": "1.2.7",
  "description": "Tamper-evident cryptographic task tracking for AI agents. Local-first: all data stays in ~/.agentpathfinder. No telemetry, no external servers, no usage caps.",
  "repository": "https://github.com/CertainLogicAI/agentpathfinder",
  "entrypoints": {
    "cli": {
      "command": [
        "python3",
        "-m",
        "agentpathfinder"
      ]
    }
  },
  "exports": {
    "task_tracking": "agentpathfinder.task_engine:TaskEngine",
    "audit_trail": "agentpathfinder.audit_trail:AuditTrail"
  },
  "tags": [
    "productivity",
    "open-source",
    "trust",
    "tamper-evident",
    "cryptography",
    "task-tracking"
  ],
  "license": "MIT",
  "author": {
    "name": "CertainLogic",
    "url": "https://github.com/CertainLogicAI"
  },
  "safety": "SAFETY.md"
}

---
name: hallucination-guard
description: "Install, configure, and use CertainLogic Verifier (hallucination‑guard) – deterministic AI verification middleware that catches hallucinations before they reach users and cuts token costs up to 98% via semantic caching. Use when: (1) setting up hallucination detection for AI agents, (2) integrating deterministic verification into LLM pipelines, (3) reducing API costs via response caching, (4) ensuring audit‑ready compliance (HIPAA/GDPR/SOC2), (5) self‑hosting AI validation layers."
---

# Hallucination Guard – CertainLogic Verifier

## Overview

CertainLogic Verifier is an open‑source, self‑hosted middleware layer that sits between your LLM calls and your application. It validates every AI response against a verified facts database, flags hallucinations, caches verified answers (bypassing the LLM), and provides cryptographic audit logs.

**Key capabilities:**
- **99%+ hallucination block rate** – rule‑based checks + TF‑IDF memory search against your `facts_db`
- **85‑98% token savings** – semantic cache hits skip the LLM entirely
- **Self‑hosted & air‑gapped** – nothing leaves your infrastructure; ready for HIPAA/GDPR/SOC2/FedRAMP
- **MIT licensed** – no proprietary lock‑in; inspect every validation rule
- **Deterministic grounding** – same query → same verified answer, every time
- **Cryptographic audit logs** – SHA‑256 chained JSONL for compliance

## Quick Start (2‑Minute Install)

```bash
# Clone the repository
git clone https://github.com/CertainLogicAI/hallucination-guard
cd hallucination-guard

# Set up Python environment
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install -r requirements.txt

# Start the service
uvicorn main:app --host 0.0.0.0 --port 8000
```

**Verify it's working:**
```bash
curl -X POST http://localhost:8000/validate \
  -d '{"query": "What is the price of GPT‑5?", "response": "\$200/month"}'
```

## Installation Options

### 1. Docker (Recommended for Production)
```bash
docker build -t hallucination-guard .
docker run -p 8000:8000 hallucination-guard
```

### 2. Kubernetes/Helm
See `deploy/helm/` in the repository for production‑ready Helm charts.

### 3. Systemd Service
A sample systemd unit file is included at `deploy/systemd/hallucination-guard.service`.

## Configuration

### Facts Database
The verifier checks responses against `facts_db.json`. Populate it with your domain‑specific verified facts.

**Example entry:**
```json
{
  "fact": "Python was created in 1991 by Guido van Rossum",
  "category": "programming",
  "source": "official Python history",
  "verified_at": "2026‑04‑20"
}
```

**Adding facts:**
- Manually edit `facts_db.json`
- Use the `/facts/add` endpoint (POST with JSON)
- Bulk‑load from documents via the `/warming/extract` endpoint

### Environment Variables
Set these in `.env` or as environment variables:

```bash
PRODUCT_MODE=coder           # coder|agent (determines rate limits)
OPENROUTER_API_KEY=your_key  # Required for cache‑miss fallback
LOG_LEVEL=INFO              # DEBUG|INFO|WARNING|ERROR
CACHE_DIR=./cache           # Persistent cache storage
```

## Usage

### Validating a Single Response
```python
import requests

response = requests.post(
    "http://localhost:8000/validate",
    json={
        "query": "What year was Python created?",
        "response": "Python was created in 1991."
    }
)
print(response.json())
```

### Integrating with AI Agent Pipelines
Place the verifier between your LLM call and your application logic:

```python
def get_ai_response(query):
    # 1. Check cache first
    cache_check = requests.post("http://localhost:8000/cache/check", 
                                json={"query": query})
    if cache_check.json().get("cached"):
        return cache_check.json()["response"]
    
    # 2. Call LLM
    llm_response = call_llm(query)
    
    # 3. Validate
    validation = requests.post("http://localhost:8000/validate",
                               json={"query": query, "response": llm_response})
    
    if validation.json().get("valid"):
        return llm_response
    else:
        # Handle hallucination
        raise ValueError(f"Hallucination detected: {validation.json()}")
```

### Cache Management
- **View cache stats:** `GET /cache/stats`
- **Clear cache:** `POST /cache/clear`
- **Warm cache:** `POST /warming/run` (requires OpenRouter API key)

## Advanced Features

### Deterministic Memory Search
The verifier uses TF‑IDF similarity to match queries against known facts, even with paraphrasing.

### Uncertainty Detection
Responses containing "I think", "might be", "not sure" are penalized and flagged for review.

### Numeric‑Unit Matching
Checks that numeric values match known facts with correct units (e.g., "5 km" vs "5 miles").

### Audit Logs
All validations are logged to `audit_log.jsonl` with SHA‑256 chaining for tamper evidence.

## Resources

### scripts/
- `install.sh` – One‑line installer for Linux/macOS
- `docker-compose.yml` – Multi‑service setup with PostgreSQL for audit logs

### references/
- `api-reference.md` – Complete API documentation
- `facts-schema.md` – Facts database schema and validation rules
- `integration-guide.md` – Step‑by‑step integration with popular AI frameworks

### assets/
- `sample-facts.json` – Example facts database with 50+ verified entries
- `docker-compose.prod.yml` – Production‑ready Docker Compose configuration

## Support & Community
- **GitHub Issues:** https://github.com/CertainLogicAI/hallucination-guard/issues
- **Discord:** https://discord.gg/clawd (OpenClaw community)
- **X/Twitter:** @CertainLogicAI

## License
MIT – see `LICENSE` in the repository.

FILE:assets/docker-compose.yml
version: '3.8'

services:
  verifier:
    build: .
    ports:
      - "8000:8000"
    environment:
      - PRODUCT_MODE=coder
      - LOG_LEVEL=INFO
      - CACHE_DIR=/app/cache
      # - OPENROUTER_API_KEY=your_key_here
    volumes:
      - ./cache:/app/cache
      - ./facts_db.json:/app/facts_db.json
      - ./audit_log.jsonl:/app/audit_log.jsonl
    restart: unless-stopped

  # Optional PostgreSQL for persistent audit logs (uncomment if needed)
  # postgres:
  #   image: postgres:15
  #   environment:
  #     POSTGRES_DB: audit
  #     POSTGRES_USER: verifier
  #     POSTGRES_PASSWORD: changeme
  #   volumes:
  #     - postgres_data:/var/lib/postgresql/data
  #   ports:
  #     - "5432:5432"

volumes:
  postgres_data:
FILE:assets/sample-facts.json
{
  "facts": {
    "2+2": {
      "type": "numeric",
      "value": "4",
      "tolerance": 0.0
    },
    "capital of france": {
      "type": "string",
      "value": "paris"
    },
    "speed of light": {
      "type": "numeric",
      "value": "299792458",
      "unit": "m/s"
    },
    "water freezes at": {
      "type": "numeric",
      "value": "0",
      "unit": "°c"
    },
    "pi": {
      "type": "numeric",
      "value": "3.1415926535"
    },
    "python release year": {
      "type": "numeric",
      "value": "1991"
    },
    "docker first release": {
      "type": "numeric",
      "value": "2013"
    },
    "http status 404": {
      "type": "string",
      "value": "not found"
    },
    "largest planet in solar system": {
      "type": "string",
      "value": "jupiter"
    },
    "einstein's theory of relativity": {
      "type": "string",
      "value": "e=mc^2"
    }
  }
}
FILE:references/api-reference.md
# API Reference – CertainLogic Verifier

Base URL: `http://localhost:8000` (or your deployment host)

All endpoints accept and return JSON.

## Validation

### `POST /validate`
Validate a query‑response pair.

**Request body:**
```json
{
  "query": "What is the price of GPT‑5?",
  "response": "$200/month"
}
```

**Response:**
```json
{
  "valid": false,
  "confidence": 0.5,
  "severity": "medium",
  "message": "Factual mismatch: No matching fact for factual query — unverifiable",
  "uncertainty_phrases": [],
  "matched_facts": []
}
```

**Fields:**
- `valid` (boolean): `true` if response passes validation
- `confidence` (float 0–1): Confidence score (higher = more certain)
- `severity` (string): `low`|`medium`|`high` (only when `valid: false`)
- `message` (string): Human‑readable explanation
- `uncertainty_phrases` (array): Phrases like `["i think", "might be"]` that triggered uncertainty penalty
- `matched_facts` (array): Verified facts that matched the response (when `valid: true`)

## Caching

### `POST /cache/check`
Check if a query is already cached.

**Request:**
```json
{
  "query": "What year was Python created?"
}
```

**Response:**
```json
{
  "cached": true,
  "response": "Python was created in 1991.",
  "confidence": 1.0,
  "source": "facts_db"
}
```

### `POST /cache/clear`
Clear the entire semantic cache.

**Response:**
```json
{
  "cleared": true,
  "entries_removed": 42
}
```

### `GET /cache/stats`
Get cache statistics.

**Response:**
```json
{
  "total_entries": 158,
  "hits": 394,
  "misses": 612,
  "hit_rate": 0.39,
  "size_mb": 0.8
}
```

## Facts Management

### `GET /facts`
List all facts in the database.

**Response:**
```json
{
  "facts": [
    {
      "fact": "Python was created in 1991 by Guido van Rossum",
      "category": "programming",
      "source": "official Python history",
      "verified_at": "2026‑04‑20"
    }
  ],
  "count": 158
}
```

### `POST /facts/add`
Add a new verified fact.

**Request:**
```json
{
  "fact": "Docker was released in 2013",
  "category": "devops",
  "source": "Docker official blog",
  "verified_at": "2026‑04‑20"
}
```

**Response:**
```json
{
  "added": true,
  "id": 159
}
```

### `DELETE /facts/{id}`
Remove a fact by its index.

## Warm‑up (Cache Pre‑loading)

### `POST /warming/extract`
Extract facts from a document (PDF, text, HTML) and add them to the facts database.

**Request:**
```json
{
  "text": "Python was created in 1991...",
  "source": "Wikipedia Python page"
}
```

**Response:**
```json
{
  "extracted": ["Python was created in 1991"],
  "added": 1
}
```

### `POST /warming/run`
Run cache warm‑up using OpenRouter API (requires `OPENROUTER_API_KEY`).

**Request:**
```json
{
  "queries": ["What is Python?", "Explain Docker"],
  "model": "anthropic/claude-haiku-4-5"
}
```

**Response:**
```json
{
  "processed": 10,
  "cached": 8,
  "skipped": 2
}
```

## Health & Metrics

### `GET /health`
Service health check.

**Response:**
```json
{
  "status": "healthy",
  "uptime_seconds": 12345,
  "version": "1.0.0"
}
```

### `GET /metrics`
Performance metrics (Prometheus‑compatible).

**Response:**
```
# HELP validation_requests_total Total validation requests
# TYPE validation_requests_total counter
validation_requests_total 1042

# HELP cache_hit_rate Cache hit rate (0‑1)
# TYPE cache_hit_rate gauge
cache_hit_rate 0.39
```
FILE:references/facts-schema.md
# Facts Database Schema

The `facts_db.json` file contains verified facts that the verifier uses to ground AI responses.

## Schema

Each fact is a JSON object with the following fields:

```json
{
  "fact": "Python was created in 1991 by Guido van Rossum",
  "category": "programming",
  "source": "official Python history",
  "verified_at": "2026‑04‑20",
  "tags": ["python", "history", "guido"],
  "confidence": 1.0
}
```

**Required fields:**
- `fact` (string): The factual statement. Should be concise, declarative, and unambiguous.
- `category` (string): Broad category for grouping (e.g., `programming`, `medical`, `financial`, `legal`).
- `source` (string): Where this fact came from (URL, document name, person).
- `verified_at` (string): Date of verification in ISO 8601 (`YYYY‑MM‑DD`).

**Optional fields:**
- `tags` (array of strings): Keywords for easier searching.
- `confidence` (number 0–1): How certain the verification is (default 1.0).
- `expires_at` (string): Date when this fact should be re‑verified (for time‑sensitive information).
- `aliases` (array of strings): Alternative phrasings of the same fact.

## Fact Quality Guidelines

### Do
- Use complete sentences: "Python was created in 1991" not "1991".
- Include units for measurements: "The speed of light is 299,792,458 m/s".
- Specify context when needed: "In the United States, the retirement age is 67".
- Add citations where possible: `source: "IRS Publication 590‑A (2024)"`.
- Keep facts atomic (one claim per fact).

### Don't
- Include subjective statements: "Python is the best programming language".
- Use vague language: "Python is quite old".
- Embed reasoning: "Because Python was created in 1991, it's older than Node.js".
- Include formatting or markdown.

## Categories

Predefined categories (you can add your own):

| Category | Description | Example |
|----------|-------------|---------|
| `programming` | Programming languages, frameworks, tools | "React was released in 2013" |
| `medical` | Medical guidelines, drug information | "Aspirin dosage is 81‑325 mg daily" |
| `financial` | Tax codes, investment rules, limits | "401(k) contribution limit for 2024 is $23,000" |
| `legal` | Laws, regulations, compliance | "GDPR requires data‑processing consent" |
| `company` | Company‑specific information | "Our support email is [email protected]" |
| `product` | Product specs, pricing, features | "Basic plan costs $29/month" |
| `historical` | Historical events, dates | "World War II ended in 1945" |
| `scientific` | Scientific constants, formulas | "E = mc²" |

## Verification Process

Before adding a fact to the database:

1. **Source check**: Verify against authoritative sources (official docs, reputable publications).
2. **Timestamp**: Record when it was verified.
3. **Conflict check**: Ensure no contradictory fact already exists.
4. **Format validation**: Ensure the fact follows schema guidelines.

## Bulk Import

You can import facts from:

- **CSV/TSV**: Use `scripts/import_facts.py` (in the repository)
- **JSON lines**: Each line a JSON object
- **Documents**: Use `/warming/extract` endpoint to extract facts from PDFs/text

## Example Database

```json
[
  {
    "fact": "Python was created in 1991 by Guido van Rossum",
    "category": "programming",
    "source": "Python.org history page",
    "verified_at": "2026‑04‑20",
    "tags": ["python", "guido", "history"]
  },
  {
    "fact": "The speed of light in vacuum is 299,792,458 meters per second",
    "category": "scientific",
    "source": "NIST CODATA 2018",
    "verified_at": "2026‑04‑20",
    "tags": ["physics", "constant"]
  },
  {
    "fact": "The standard deduction for single filers in 2024 is $14,600",
    "category": "financial",
    "source": "IRS Revenue Procedure 2023‑34",
    "verified_at": "2026‑04‑20",
    "expires_at": "2025‑01‑01",
    "tags": ["tax", "irs", "2024"]
  }
]
```
FILE:references/integration-guide.md
# Integration Guide

How to integrate CertainLogic Verifier with popular AI frameworks and agent platforms.

## OpenClaw / Codex

Add the verifier as a middleware step in your OpenClaw skill or agent:

```python
# In your skill's SKILL.md or agent logic
import requests

def validate_response(query, response):
    """Validate an AI response using the local verifier."""
    result = requests.post(
        "http://localhost:8000/validate",
        json={"query": query, "response": response},
        timeout=5
    ).json()
    
    if not result.get("valid"):
        # Flag for human review or retry
        return {
            "safe": False,
            "confidence": result["confidence"],
            "reason": result["message"]
        }
    return {"safe": True, "confidence": result["confidence"]}
```

**Configuration in OpenClaw:** Set the verifier URL as an environment variable:
```bash
export HALLUCINATION_GUARD_URL=http://localhost:8000
```

## LangChain

Create a custom `HallucinationGuard` output parser:

```python
from langchain.output_parsers import BaseOutputParser
import requests

class HallucinationGuardParser(BaseOutputParser):
    def parse(self, text: str, query: str = None):
        response = requests.post(
            "http://localhost:8000/validate",
            json={"query": query, "response": text}
        ).json()
        
        if not response["valid"]:
            raise ValueError(f"Hallucination detected: {response['message']}")
        return text

# Usage
parser = HallucinationGuardParser()
chain = LLMChain(llm=llm, prompt=prompt, output_parser=parser)
```

## OpenAI / Anthropic API Wrapper

Wrap the API call with validation:

```python
import openai
import requests

def safe_completion(prompt, model="gpt-4"):
    # 1. Get LLM response
    response = openai.ChatCompletion.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    ).choices[0].message.content
    
    # 2. Validate
    validation = requests.post(
        "http://localhost:8000/validate",
        json={"query": prompt, "response": response}
    ).json()
    
    if not validation["valid"]:
        # Fallback: return a safe message or escalate
        return f"[VALIDATION FAILED] {validation['message']}"
    
    return response
```

## FastAPI / Flask Middleware

Add the verifier as middleware in your web app:

```python
from fastapi import FastAPI, Request
import requests

app = FastAPI()

@app.middleware("http")
async def validate_ai_responses(request: Request, call_next):
    response = await call_next(request)
    
    # Only validate AI endpoints
    if request.url.path == "/ai/chat":
        body = await request.json()
        validation = requests.post(
            "http://localhost:8000/validate",
            json={"query": body["query"], "response": body["response"]}
        )
        # Add validation header
        response.headers["X-Validation-Status"] = validation.json()["valid"]
    
    return response
```

## Docker Compose Integration

Add the verifier to your existing Docker Compose stack:

```yaml
version: '3.8'
services:
  your-app:
    # ... your existing service
  
  hallucination-guard:
    image: certainlogic/hallucination-guard:latest
    ports:
      - "8000:8000"
    environment:
      - PRODUCT_MODE=agent
    volumes:
      - ./facts_db.json:/app/facts_db.json
```

## Kubernetes Sidecar

Deploy the verifier as a sidecar container:

```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-agent
spec:
  template:
    spec:
      containers:
      - name: agent
        image: your-ai-agent:latest
      - name: hallucination-guard
        image: certainlogic/hallucination-guard:latest
        ports:
        - containerPort: 8000
        env:
        - name: PRODUCT_MODE
          value: "agent"
```

## Monitoring & Alerting

Set up alerts when hallucination rate exceeds threshold:

```python
# Periodic monitoring script
import requests
from datetime import datetime

stats = requests.get("http://localhost:8000/metrics").text
# Parse Prometheus metrics
lines = stats.split('\\n')
for line in lines:
    if 'validation_requests_total' in line:
        total = int(line.split()[1])
    if 'cache_hit_rate' in line:
        hit_rate = float(line.split()[1])

if hit_rate < 0.3:  # Alert if cache hit rate drops
    send_alert(f"Cache hit rate low: {hit_rate}")
```

## Performance Considerations

- **Latency**: Validation adds ~50‑100ms per request (mostly network overhead).
- **Throughput**: Single instance handles ~100 req/s; scale horizontally for higher load.
- **Cache size**: Memory usage scales with cached entries (~1MB per 10,000 entries).

## Fallback Strategies

When the verifier is unavailable:

```python
def get_validated_response(query):
    try:
        return call_with_validation(query)
    except ConnectionError:
        # Fallback to direct LLM call with warning
        logger.warning("Validation service down, proceeding without checks")
        return call_llm_directly(query)
```
FILE:scripts/install.sh
#!/bin/bash
# One‑line installer for CertainLogic Verifier (hallucination‑guard)
# Usage: curl -sSL https://raw.githubusercontent.com/CertainLogicAI/hallucination-guard/main/install.sh | bash

set -e

REPO="https://github.com/CertainLogicAI/hallucination-guard.git"
TARGET_DIR="-./hallucination-guard"

echo "📦 Installing CertainLogic Verifier..."

# Clone repository
if [ -d "$TARGET_DIR" ]; then
    echo "⚠️  Directory $TARGET_DIR already exists. Pulling latest changes..."
    cd "$TARGET_DIR"
    git pull origin main
else
    git clone "$REPO" "$TARGET_DIR"
    cd "$TARGET_DIR"
fi

# Create virtual environment
python3 -m venv venv
source venv/bin/activate

# Install dependencies
pip install -r requirements.txt

# Create default .env file if not present
if [ ! -f .env ]; then
    cat > .env << EOF
PRODUCT_MODE=coder
LOG_LEVEL=INFO
CACHE_DIR=./cache
# OPENROUTER_API_KEY=your_key_here  # Uncomment for cache‑miss fallback
EOF
    echo "✅ Created .env file with default settings"
fi

echo "✅ Installation complete!"
echo ""
echo "To start the service:"
echo "  cd $TARGET_DIR"
echo "  source venv/bin/activate"
echo "  uvicorn main:app --host 0.0.0.0 --port 8000"
echo ""
echo "Quick test:"
echo "  curl -X POST http://localhost:8000/validate \\"
echo "    -d '{\"query\": \"Price of GPT‑5?\", \"response\": \"\\\$200/month\"}'"
echo ""
echo "See README.md for advanced configuration and integration."

---
name: hallucination-guard
description: "Install, configure, and use CertainLogic Verifier (hallucination‑guard) – deterministic AI verification middleware that catches hallucinations before they reach users and cuts token costs up to 98% via semantic caching. Use when: (1) setting up hallucination detection for AI agents, (2) integrating deterministic verification into LLM pipelines, (3) reducing API costs via response caching, (4) ensuring audit‑ready compliance (HIPAA/GDPR/SOC2), (5) self‑hosting AI validation layers."
---

# Hallucination Guard – CertainLogic Verifier

## Overview

CertainLogic Verifier is an open‑source, self‑hosted middleware layer that sits between your LLM calls and your application. It validates every AI response against a verified facts database, flags hallucinations, caches verified answers (bypassing the LLM), and provides cryptographic audit logs.

**Key capabilities:**
- **99%+ hallucination block rate** – rule‑based checks + TF‑IDF memory search against your `facts_db`
- **85‑98% token savings** – semantic cache hits skip the LLM entirely
- **Self‑hosted & air‑gapped** – nothing leaves your infrastructure; ready for HIPAA/GDPR/SOC2/FedRAMP
- **MIT licensed** – no proprietary lock‑in; inspect every validation rule
- **Deterministic grounding** – same query → same verified answer, every time
- **Cryptographic audit logs** – SHA‑256 chained JSONL for compliance

## Quick Start (2‑Minute Install)

```bash
# Clone the repository
git clone https://github.com/CertainLogicAI/hallucination-guard
cd hallucination-guard

# Set up Python environment
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install -r requirements.txt

# Start the service
uvicorn main:app --host 0.0.0.0 --port 8000
```

**Verify it's working:**
```bash
curl -X POST http://localhost:8000/validate \
  -d '{"query": "What is the price of GPT‑5?", "response": "\$200/month"}'
```

## Installation Options

### 1. Docker (Recommended for Production)
```bash
docker build -t hallucination-guard .
docker run -p 8000:8000 hallucination-guard
```

### 2. Kubernetes/Helm
See `deploy/helm/` in the repository for production‑ready Helm charts.

### 3. Systemd Service
A sample systemd unit file is included at `deploy/systemd/hallucination-guard.service`.

## Configuration

### Facts Database
The verifier checks responses against `facts_db.json`. Populate it with your domain‑specific verified facts.

**Example entry:**
```json
{
  "fact": "Python was created in 1991 by Guido van Rossum",
  "category": "programming",
  "source": "official Python history",
  "verified_at": "2026‑04‑20"
}
```

**Adding facts:**
- Manually edit `facts_db.json`
- Use the `/facts/add` endpoint (POST with JSON)
- Bulk‑load from documents via the `/warming/extract` endpoint

### Environment Variables
Set these in `.env` or as environment variables:

```bash
PRODUCT_MODE=coder           # coder|agent (determines rate limits)
OPENROUTER_API_KEY=your_key  # Required for cache‑miss fallback
LOG_LEVEL=INFO              # DEBUG|INFO|WARNING|ERROR
CACHE_DIR=./cache           # Persistent cache storage
```

## Usage

### Validating a Single Response
```python
import requests

response = requests.post(
    "http://localhost:8000/validate",
    json={
        "query": "What year was Python created?",
        "response": "Python was created in 1991."
    }
)
print(response.json())
```

### Integrating with AI Agent Pipelines
Place the verifier between your LLM call and your application logic:

```python
def get_ai_response(query):
    # 1. Check cache first
    cache_check = requests.post("http://localhost:8000/cache/check", 
                                json={"query": query})
    if cache_check.json().get("cached"):
        return cache_check.json()["response"]
    
    # 2. Call LLM
    llm_response = call_llm(query)
    
    # 3. Validate
    validation = requests.post("http://localhost:8000/validate",
                               json={"query": query, "response": llm_response})
    
    if validation.json().get("valid"):
        return llm_response
    else:
        # Handle hallucination
        raise ValueError(f"Hallucination detected: {validation.json()}")
```

### Cache Management
- **View cache stats:** `GET /cache/stats`
- **Clear cache:** `POST /cache/clear`
- **Warm cache:** `POST /warming/run` (requires OpenRouter API key)

## Advanced Features

### Deterministic Memory Search
The verifier uses TF‑IDF similarity to match queries against known facts, even with paraphrasing.

### Uncertainty Detection
Responses containing "I think", "might be", "not sure" are penalized and flagged for review.

### Numeric‑Unit Matching
Checks that numeric values match known facts with correct units (e.g., "5 km" vs "5 miles").

### Audit Logs
All validations are logged to `audit_log.jsonl` with SHA‑256 chaining for tamper evidence.

## Resources

### scripts/
- `install.sh` – One‑line installer for Linux/macOS
- `docker-compose.yml` – Multi‑service setup with PostgreSQL for audit logs

### references/
- `api-reference.md` – Complete API documentation
- `facts-schema.md` – Facts database schema and validation rules
- `integration-guide.md` – Step‑by‑step integration with popular AI frameworks

### assets/
- `sample-facts.json` – Example facts database with 50+ verified entries
- `docker-compose.prod.yml` – Production‑ready Docker Compose configuration

## Support & Community
- **GitHub Issues:** https://github.com/CertainLogicAI/hallucination-guard/issues
- **Discord:** https://discord.gg/clawd (OpenClaw community)
- **X/Twitter:** @CertainLogicAI

## License
MIT – see `LICENSE` in the repository.

FILE:assets/docker-compose.yml
version: '3.8'

services:
  verifier:
    build: .
    ports:
      - "8000:8000"
    environment:
      - PRODUCT_MODE=coder
      - LOG_LEVEL=INFO
      - CACHE_DIR=/app/cache
      # - OPENROUTER_API_KEY=your_key_here
    volumes:
      - ./cache:/app/cache
      - ./facts_db.json:/app/facts_db.json
      - ./audit_log.jsonl:/app/audit_log.jsonl
    restart: unless-stopped

  # Optional PostgreSQL for persistent audit logs (uncomment if needed)
  # postgres:
  #   image: postgres:15
  #   environment:
  #     POSTGRES_DB: audit
  #     POSTGRES_USER: verifier
  #     POSTGRES_PASSWORD: changeme
  #   volumes:
  #     - postgres_data:/var/lib/postgresql/data
  #   ports:
  #     - "5432:5432"

volumes:
  postgres_data:
FILE:assets/sample-facts.json
{
  "facts": {
    "2+2": {
      "type": "numeric",
      "value": "4",
      "tolerance": 0.0
    },
    "capital of france": {
      "type": "string",
      "value": "paris"
    },
    "speed of light": {
      "type": "numeric",
      "value": "299792458",
      "unit": "m/s"
    },
    "water freezes at": {
      "type": "numeric",
      "value": "0",
      "unit": "°c"
    },
    "pi": {
      "type": "numeric",
      "value": "3.1415926535"
    },
    "python release year": {
      "type": "numeric",
      "value": "1991"
    },
    "docker first release": {
      "type": "numeric",
      "value": "2013"
    },
    "http status 404": {
      "type": "string",
      "value": "not found"
    },
    "largest planet in solar system": {
      "type": "string",
      "value": "jupiter"
    },
    "einstein's theory of relativity": {
      "type": "string",
      "value": "e=mc^2"
    }
  }
}
FILE:references/api-reference.md
# API Reference – CertainLogic Verifier

Base URL: `http://localhost:8000` (or your deployment host)

All endpoints accept and return JSON.

## Validation

### `POST /validate`
Validate a query‑response pair.

**Request body:**
```json
{
  "query": "What is the price of GPT‑5?",
  "response": "$200/month"
}
```

**Response:**
```json
{
  "valid": false,
  "confidence": 0.5,
  "severity": "medium",
  "message": "Factual mismatch: No matching fact for factual query — unverifiable",
  "uncertainty_phrases": [],
  "matched_facts": []
}
```

**Fields:**
- `valid` (boolean): `true` if response passes validation
- `confidence` (float 0–1): Confidence score (higher = more certain)
- `severity` (string): `low`|`medium`|`high` (only when `valid: false`)
- `message` (string): Human‑readable explanation
- `uncertainty_phrases` (array): Phrases like `["i think", "might be"]` that triggered uncertainty penalty
- `matched_facts` (array): Verified facts that matched the response (when `valid: true`)

## Caching

### `POST /cache/check`
Check if a query is already cached.

**Request:**
```json
{
  "query": "What year was Python created?"
}
```

**Response:**
```json
{
  "cached": true,
  "response": "Python was created in 1991.",
  "confidence": 1.0,
  "source": "facts_db"
}
```

### `POST /cache/clear`
Clear the entire semantic cache.

**Response:**
```json
{
  "cleared": true,
  "entries_removed": 42
}
```

### `GET /cache/stats`
Get cache statistics.

**Response:**
```json
{
  "total_entries": 158,
  "hits": 394,
  "misses": 612,
  "hit_rate": 0.39,
  "size_mb": 0.8
}
```

## Facts Management

### `GET /facts`
List all facts in the database.

**Response:**
```json
{
  "facts": [
    {
      "fact": "Python was created in 1991 by Guido van Rossum",
      "category": "programming",
      "source": "official Python history",
      "verified_at": "2026‑04‑20"
    }
  ],
  "count": 158
}
```

### `POST /facts/add`
Add a new verified fact.

**Request:**
```json
{
  "fact": "Docker was released in 2013",
  "category": "devops",
  "source": "Docker official blog",
  "verified_at": "2026‑04‑20"
}
```

**Response:**
```json
{
  "added": true,
  "id": 159
}
```

### `DELETE /facts/{id}`
Remove a fact by its index.

## Warm‑up (Cache Pre‑loading)

### `POST /warming/extract`
Extract facts from a document (PDF, text, HTML) and add them to the facts database.

**Request:**
```json
{
  "text": "Python was created in 1991...",
  "source": "Wikipedia Python page"
}
```

**Response:**
```json
{
  "extracted": ["Python was created in 1991"],
  "added": 1
}
```

### `POST /warming/run`
Run cache warm‑up using OpenRouter API (requires `OPENROUTER_API_KEY`).

**Request:**
```json
{
  "queries": ["What is Python?", "Explain Docker"],
  "model": "anthropic/claude-haiku-4-5"
}
```

**Response:**
```json
{
  "processed": 10,
  "cached": 8,
  "skipped": 2
}
```

## Health & Metrics

### `GET /health`
Service health check.

**Response:**
```json
{
  "status": "healthy",
  "uptime_seconds": 12345,
  "version": "1.0.0"
}
```

### `GET /metrics`
Performance metrics (Prometheus‑compatible).

**Response:**
```
# HELP validation_requests_total Total validation requests
# TYPE validation_requests_total counter
validation_requests_total 1042

# HELP cache_hit_rate Cache hit rate (0‑1)
# TYPE cache_hit_rate gauge
cache_hit_rate 0.39
```
FILE:references/facts-schema.md
# Facts Database Schema

The `facts_db.json` file contains verified facts that the verifier uses to ground AI responses.

## Schema

Each fact is a JSON object with the following fields:

```json
{
  "fact": "Python was created in 1991 by Guido van Rossum",
  "category": "programming",
  "source": "official Python history",
  "verified_at": "2026‑04‑20",
  "tags": ["python", "history", "guido"],
  "confidence": 1.0
}
```

**Required fields:**
- `fact` (string): The factual statement. Should be concise, declarative, and unambiguous.
- `category` (string): Broad category for grouping (e.g., `programming`, `medical`, `financial`, `legal`).
- `source` (string): Where this fact came from (URL, document name, person).
- `verified_at` (string): Date of verification in ISO 8601 (`YYYY‑MM‑DD`).

**Optional fields:**
- `tags` (array of strings): Keywords for easier searching.
- `confidence` (number 0–1): How certain the verification is (default 1.0).
- `expires_at` (string): Date when this fact should be re‑verified (for time‑sensitive information).
- `aliases` (array of strings): Alternative phrasings of the same fact.

## Fact Quality Guidelines

### Do
- Use complete sentences: "Python was created in 1991" not "1991".
- Include units for measurements: "The speed of light is 299,792,458 m/s".
- Specify context when needed: "In the United States, the retirement age is 67".
- Add citations where possible: `source: "IRS Publication 590‑A (2024)"`.
- Keep facts atomic (one claim per fact).

### Don't
- Include subjective statements: "Python is the best programming language".
- Use vague language: "Python is quite old".
- Embed reasoning: "Because Python was created in 1991, it's older than Node.js".
- Include formatting or markdown.

## Categories

Predefined categories (you can add your own):

| Category | Description | Example |
|----------|-------------|---------|
| `programming` | Programming languages, frameworks, tools | "React was released in 2013" |
| `medical` | Medical guidelines, drug information | "Aspirin dosage is 81‑325 mg daily" |
| `financial` | Tax codes, investment rules, limits | "401(k) contribution limit for 2024 is $23,000" |
| `legal` | Laws, regulations, compliance | "GDPR requires data‑processing consent" |
| `company` | Company‑specific information | "Our support email is [email protected]" |
| `product` | Product specs, pricing, features | "Basic plan costs $29/month" |
| `historical` | Historical events, dates | "World War II ended in 1945" |
| `scientific` | Scientific constants, formulas | "E = mc²" |

## Verification Process

Before adding a fact to the database:

1. **Source check**: Verify against authoritative sources (official docs, reputable publications).
2. **Timestamp**: Record when it was verified.
3. **Conflict check**: Ensure no contradictory fact already exists.
4. **Format validation**: Ensure the fact follows schema guidelines.

## Bulk Import

You can import facts from:

- **CSV/TSV**: Use `scripts/import_facts.py` (in the repository)
- **JSON lines**: Each line a JSON object
- **Documents**: Use `/warming/extract` endpoint to extract facts from PDFs/text

## Example Database

```json
[
  {
    "fact": "Python was created in 1991 by Guido van Rossum",
    "category": "programming",
    "source": "Python.org history page",
    "verified_at": "2026‑04‑20",
    "tags": ["python", "guido", "history"]
  },
  {
    "fact": "The speed of light in vacuum is 299,792,458 meters per second",
    "category": "scientific",
    "source": "NIST CODATA 2018",
    "verified_at": "2026‑04‑20",
    "tags": ["physics", "constant"]
  },
  {
    "fact": "The standard deduction for single filers in 2024 is $14,600",
    "category": "financial",
    "source": "IRS Revenue Procedure 2023‑34",
    "verified_at": "2026‑04‑20",
    "expires_at": "2025‑01‑01",
    "tags": ["tax", "irs", "2024"]
  }
]
```
FILE:references/integration-guide.md
# Integration Guide

How to integrate CertainLogic Verifier with popular AI frameworks and agent platforms.

## OpenClaw / Codex

Add the verifier as a middleware step in your OpenClaw skill or agent:

```python
# In your skill's SKILL.md or agent logic
import requests

def validate_response(query, response):
    """Validate an AI response using the local verifier."""
    result = requests.post(
        "http://localhost:8000/validate",
        json={"query": query, "response": response},
        timeout=5
    ).json()
    
    if not result.get("valid"):
        # Flag for human review or retry
        return {
            "safe": False,
            "confidence": result["confidence"],
            "reason": result["message"]
        }
    return {"safe": True, "confidence": result["confidence"]}
```

**Configuration in OpenClaw:** Set the verifier URL as an environment variable:
```bash
export HALLUCINATION_GUARD_URL=http://localhost:8000
```

## LangChain

Create a custom `HallucinationGuard` output parser:

```python
from langchain.output_parsers import BaseOutputParser
import requests

class HallucinationGuardParser(BaseOutputParser):
    def parse(self, text: str, query: str = None):
        response = requests.post(
            "http://localhost:8000/validate",
            json={"query": query, "response": text}
        ).json()
        
        if not response["valid"]:
            raise ValueError(f"Hallucination detected: {response['message']}")
        return text

# Usage
parser = HallucinationGuardParser()
chain = LLMChain(llm=llm, prompt=prompt, output_parser=parser)
```

## OpenAI / Anthropic API Wrapper

Wrap the API call with validation:

```python
import openai
import requests

def safe_completion(prompt, model="gpt-4"):
    # 1. Get LLM response
    response = openai.ChatCompletion.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    ).choices[0].message.content
    
    # 2. Validate
    validation = requests.post(
        "http://localhost:8000/validate",
        json={"query": prompt, "response": response}
    ).json()
    
    if not validation["valid"]:
        # Fallback: return a safe message or escalate
        return f"[VALIDATION FAILED] {validation['message']}"
    
    return response
```

## FastAPI / Flask Middleware

Add the verifier as middleware in your web app:

```python
from fastapi import FastAPI, Request
import requests

app = FastAPI()

@app.middleware("http")
async def validate_ai_responses(request: Request, call_next):
    response = await call_next(request)
    
    # Only validate AI endpoints
    if request.url.path == "/ai/chat":
        body = await request.json()
        validation = requests.post(
            "http://localhost:8000/validate",
            json={"query": body["query"], "response": body["response"]}
        )
        # Add validation header
        response.headers["X-Validation-Status"] = validation.json()["valid"]
    
    return response
```

## Docker Compose Integration

Add the verifier to your existing Docker Compose stack:

```yaml
version: '3.8'
services:
  your-app:
    # ... your existing service
  
  hallucination-guard:
    image: certainlogic/hallucination-guard:latest
    ports:
      - "8000:8000"
    environment:
      - PRODUCT_MODE=agent
    volumes:
      - ./facts_db.json:/app/facts_db.json
```

## Kubernetes Sidecar

Deploy the verifier as a sidecar container:

```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-agent
spec:
  template:
    spec:
      containers:
      - name: agent
        image: your-ai-agent:latest
      - name: hallucination-guard
        image: certainlogic/hallucination-guard:latest
        ports:
        - containerPort: 8000
        env:
        - name: PRODUCT_MODE
          value: "agent"
```

## Monitoring & Alerting

Set up alerts when hallucination rate exceeds threshold:

```python
# Periodic monitoring script
import requests
from datetime import datetime

stats = requests.get("http://localhost:8000/metrics").text
# Parse Prometheus metrics
lines = stats.split('\\n')
for line in lines:
    if 'validation_requests_total' in line:
        total = int(line.split()[1])
    if 'cache_hit_rate' in line:
        hit_rate = float(line.split()[1])

if hit_rate < 0.3:  # Alert if cache hit rate drops
    send_alert(f"Cache hit rate low: {hit_rate}")
```

## Performance Considerations

- **Latency**: Validation adds ~50‑100ms per request (mostly network overhead).
- **Throughput**: Single instance handles ~100 req/s; scale horizontally for higher load.
- **Cache size**: Memory usage scales with cached entries (~1MB per 10,000 entries).

## Fallback Strategies

When the verifier is unavailable:

```python
def get_validated_response(query):
    try:
        return call_with_validation(query)
    except ConnectionError:
        # Fallback to direct LLM call with warning
        logger.warning("Validation service down, proceeding without checks")
        return call_llm_directly(query)
```
FILE:scripts/install.sh
#!/bin/bash
# One‑line installer for CertainLogic Verifier (hallucination‑guard)
# Usage: curl -sSL https://raw.githubusercontent.com/CertainLogicAI/hallucination-guard/main/install.sh | bash

set -e

REPO="https://github.com/CertainLogicAI/hallucination-guard.git"
TARGET_DIR="-./hallucination-guard"

echo "📦 Installing CertainLogic Verifier..."

# Clone repository
if [ -d "$TARGET_DIR" ]; then
    echo "⚠️  Directory $TARGET_DIR already exists. Pulling latest changes..."
    cd "$TARGET_DIR"
    git pull origin main
else
    git clone "$REPO" "$TARGET_DIR"
    cd "$TARGET_DIR"
fi

# Create virtual environment
python3 -m venv venv
source venv/bin/activate

# Install dependencies
pip install -r requirements.txt

# Create default .env file if not present
if [ ! -f .env ]; then
    cat > .env << EOF
PRODUCT_MODE=coder
LOG_LEVEL=INFO
CACHE_DIR=./cache
# OPENROUTER_API_KEY=your_key_here  # Uncomment for cache‑miss fallback
EOF
    echo "✅ Created .env file with default settings"
fi

echo "✅ Installation complete!"
echo ""
echo "To start the service:"
echo "  cd $TARGET_DIR"
echo "  source venv/bin/activate"
echo "  uvicorn main:app --host 0.0.0.0 --port 8000"
echo ""
echo "Quick test:"
echo "  curl -X POST http://localhost:8000/validate \\"
echo "    -d '{\"query\": \"Price of GPT‑5?\", \"response\": \"\\\$200/month\"}'"
echo ""
echo "See README.md for advanced configuration and integration."

---
name: certainlogic-context-manager
license: MIT
description: Prevent AI session token bloat and runaway costs. Tracks query count per session, warns at a configurable threshold, auto-writes a dense handoff summary before reset, and reads it back on the next session start — so no work is lost and every session starts lean. Without session resets, a standard OpenClaw session can balloon from 18K to 400K+ input tokens, multiplying cost 22x. This skill prevents that automatically. Use when managing long AI sessions, reducing token burn, keeping context lean, or implementing session hygiene for any OpenClaw-based agent. Triggers on phrases like "context management", "session reset", "token bloat", "query counter", "handoff", "save context", or "/new". Built by CertainLogic.
---

# Context Manager

Keeps sessions lean. Tracks query count, warns before context bloats, saves a handoff summary so work continues cleanly in the next session.

## How It Works

1. **Counter** — Increment `session_query_count.txt` every turn
2. **Warn** — At threshold, write `handoff.md` and alert the user
3. **Reset** — User runs `/new`; next session reads and deletes `handoff.md`

## Files

| File | Purpose |
|---|---|
| `session_query_count.txt` | Integer. Increment each turn. Create if missing (default `0`). |
| `handoff.md` | Workspace root. Written at threshold. Read + deleted on next session start. |

## Every Turn (mandatory)

```
1. Read session_query_count.txt (default 0 if missing)
2. Increment by 1
3. Write back
4. If count == THRESHOLD: trigger handoff flow
```

Default threshold: **10**. Configurable per deployment.

## Handoff Flow (at threshold)

1. Write `handoff.md` — see [handoff-format.md](references/handoff-format.md) for exact spec
2. Tell the user:
   > ⚠️ **10 queries in** — context is getting heavy. Handoff saved. Run `/new` when ready to continue.
3. Do NOT auto-reset. User controls the reset.

## Session Start Flow

```
1. Check if handoff.md exists
2. If exists AND file is < 3 hours old:
   a. Read it
   b. Apply context to current session
   c. Delete handoff.md
3. Reset session_query_count.txt to 0
```

## Topic Switch Detection

When the user says **"BTW"**, "switching gears", "new topic", or starts a clearly unrelated task:

1. Write/overwrite `handoff.md` immediately (don't wait for threshold)
2. Tell the user: *"Handoff saved — `/new` when ready"*
3. Resume or continue current task

## Why This Matters

Session context is cumulative. Every prior message, tool call, and assistant response is re-sent on every new query. Without resets, a lean 18K-token session becomes 50K–400K+ within hours.

See [token-math.md](references/token-math.md) for cost impact data and reset savings estimates.

## Commands

| Command | Behavior |
|---|---|
| `/new` or `/reset` | User-triggered. New session starts; skill reads handoff.md if present. |
| `/handoff` | Explicit handoff write. Same as threshold flow, on demand. |
| `/counter` | Report current query count and threshold. |

## Configuration

Override defaults in `AGENTS.md` or equivalent workspace config:

```
- CONTEXT_THRESHOLD: 10        # Queries before warning (default: 10)
- HANDOFF_TTL_HOURS: 3         # Hours before handoff.md is considered stale (default: 3)
```

## Reliability Rules

- **Never skip the counter increment** — even for one-liner responses
- **Never auto-reset** — the user owns the session lifecycle
- **Handoff must be written before warning** — never warn without saving state first
- **Counter file is source of truth** — do not track in memory

---

## What This Saves You

| Without this skill | With this skill |
|---|---|
| Sessions balloon to 400K+ tokens | Sessions stay under 25K |
| Cost multiplies 22x over a day | Cost stays near baseline |
| Work lost on manual /new resets | Handoff auto-saved, resumed next session |
| No signal before context degrades | Warning fires before quality drops |

See [token-math.md](references/token-math.md) for the full cost breakdown.

---

*Built by [CertainLogic](https://certainlogic.ai) — trusted fact infrastructure for AI agents.*

FILE:references/handoff-format.md
# Handoff Format Specification

Handoffs are dense summaries written at session boundaries so the next session can resume without re-reading the full conversation history.

## File Location
`handoff.md` in the workspace root. Overwrite on each new handoff — only the latest matters.

## Format

```markdown
# Session Handoff
**Generated:** [YYYY-MM-DD HH:MM] [timezone]

## Current Task
[What was being worked on — 1–2 sentences max]

## Last Decision
[The last significant thing decided, completed, or changed]

## Open Items
- [Unfinished work or pending decisions]
- [Each item on its own line]

## Key Context
[1–3 facts the next session needs to avoid repeating work or making wrong assumptions]
```

## Rules
- **Max 20 lines.** Dense, not verbose.
- **No pleasantries.** Write for a cold-start agent, not a human reader.
- **Tense:** Past tense for completed work. Present tense for open items.
- **No opinions or summaries of the conversation.** Only actionable facts.

## When to Write a Handoff
- Query counter hits the configured threshold (default: 10)
- User says "BTW", "switching gears", "new topic", or starts a clearly unrelated task
- User explicitly requests `/handoff`

## When to Read a Handoff
- At session start — check if `handoff.md` exists and is < 3 hours old
- If recent: read, apply context, then delete the file

## Example

```markdown
# Session Handoff
**Generated:** 2026-04-16 09:30 CDT

## Current Task
Building context-manager skill for CertainLogic bundle.

## Last Decision
Decided on query-count (not timer) as primary bloat signal. 10-query default.

## Open Items
- Write SKILL.md body
- Package and test .skill file
- Add to ClawHub listing

## Key Context
- Skill is free, bundled with paid CertainLogic products
- Must reflect CertainLogic quality bar — it's a product advertisement
- Token math reference saved to references/token-math.md
```

FILE:references/token-math.md
# Token Math Reference

## Why Sessions Bloat

Every message in a session accumulates in the context window. OpenClaw loads:
- System prompt + workspace files (~18K tokens baseline for a typical setup)
- All prior assistant turns
- All prior user turns
- Tool call inputs and outputs

Each query adds ~1K–5K tokens depending on message length and tool use. After 10 queries, a lean session (~18K baseline) can reach 50K–80K tokens. After 20+, it can exceed 400K.

## Cost Impact (Claude Sonnet 4.x pricing reference)

| Tokens | Est. cost per query | Notes |
|---|---|---|
| 18K (new session) | ~$0.054 | Baseline — fresh /new |
| 50K (10 queries) | ~$0.150 | 2.8x baseline |
| 100K (20+ queries) | ~$0.300 | 5.5x baseline |
| 400K (bloated) | ~$1.200 | 22x baseline |

*Costs are estimates. Cached tokens billed at lower rates where applicable.*

## The Compounding Problem

Each new message pays for ALL prior context — every assistant turn, every tool call.
Session cost is O(n²) in message count without resets.

## Reset Saves

Resetting at 10 queries vs. continuing to 20 saves approximately:
- 50% of input tokens for the second 10 queries
- Compounds across the full day if multiple sessions run

## Session Baseline by Setup Size

| Workspace size | Typical baseline |
|---|---|
| Minimal (no project files) | ~5K–8K |
| Standard (SOUL + USER + MEMORY) | ~12K–20K |
| Heavy (large MEMORY + daily logs) | ~25K–40K |

## Signals That a Session Is Bloated

- Responses feel slower than usual
- The agent references stale context from early in the session
- Cost per query visibly increasing
- Tool call outputs are large and repeated (e.g., file reads cached in context)