# Project export: IMMUNE This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: UC Berkeley AI Hackathon 2026 - Tagline: AI agents trust their memory blindly, one poisoned note corrupts every future answer. IMMUNE proves the guilty memory by deterministic replay, quarantines it in Redis, and heals the agent. - Devpost: https://devpost.com/software/immunify-dnv043 - GitHub: https://github.com/Sixxhi/AI-Hackathon-20206 - Video: https://www.youtube.com/embed/MIYsgGiZ5k8?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 6 GitHub contributor(s) — Claude Opus 4.8 (1M context) (56 commits), Vineeth Reddy Vallapureddy (34 commits), sidd-hi (16 commits), Vineeth Reddy Vallapureddy (15 commits), Saanvi Bhargava (14 commits), sszz01 (8 commits) ## Devpost submission (written by the team) ### Inspiration Agents have long term memory now, and they trust whatever they pull out of it. That turns out to be a real hole. If a bad fact gets written into that memory, the agent retrieves it later and acts on it without a second thought. OWASP gave this its own category in 2026 (ASI06, memory and context poisoning), and it has already been shown working against ChatGPT, Gemini, and Bedrock. What got us going was a smaller observation. Most defenses ask a second model whether a memory is trustworthy, and if your judge is a language model, the same poisoning that fools the agent can fool the judge. We wanted to point at the exact memory that caused a wrong answer and prove it, with no model involved in that call. ### What it does IMMUNE sits between an agent and its memory and watches both directions. On the way in, every memory gets tagged with where it came from, and that provenance sets its trust. Something a user typed cannot quietly outrank the official policy. On the way out, recall runs through Redis vector search, and anything that has been quarantined is filtered out by the index itself, so the agent never even sees it. The interesting part happens after a wrong answer. When the agent says something that conflicts with a memory we trust more, IMMUNE re runs that same question with different memories pulled out, and watches for the moment the answer flips back to correct. The memory whose removal fixes things is the culprit. We quarantine it, log an incident to Sentry, and let the agent answer again, now correctly. No language model is involved in deciding which memory was at fault.. ### How we built it We chose to use Claude Haiku 4.5 as our model choice through the Anthropic API. Its memory lives in Redis with RediSearch, using a FLAT index and cosine distance. Each memory is stored as a hash with a float32 embedding, and recall is a k-nearest-neighbor query. Quarantine is part of the query itself through a status filter, which is why a jailed memory simply cannot come back. The attribution engine is the piece we spent the most time on. It removes memories in an adaptive order, suspicious ones first, then runs a delta debugging pass to shrink the result down to the smallest set of memories that actually caused the failure. Detection works without any ground truth, because it only fires when an answer disagrees with a higher trust memory, which is the situation you actually have in production. Trust comes entirely from provenance and from a memory's own history (official documents start at 0.9, user writes at 0.5, trust collapses on quarantine and recovers on parole). We kept it completely separate from similarity on purpose. Around that core we wired Sentry for incident reporting, Arize Phoenix and OpenTelemetry for tracing, a LangGraph store adapter, an MCP server for Claude Code, and a Streamlit dashboard. There are 44 tests, and the whole thing also runs fully offline and deterministic as a backup. Sentry is used as an alerting layer for the memory firewall: when the quarantine() function detects a poisoned or malicious memory record, it fires a sentry_sdk.capture_message event (via _sentry_quarantine in immune/store.py) tagged with the threat type and offending content. It's initialized lazily via SENTRY_DSN/USE_SENTRY env vars and is a no-op when unconfigured, so the core firewall logic runs without it. Link to our Sentry org: https://berkeley-hackathon2026.sentry.io/. ### Challenges we ran into Throughout our build, IMMUNE surfaced challenges at every layer. The problem space is open-ended and unsolved: memory integrity spans poisoning, rot, and pollution, and scoping what to demo first was difficult in itself. From there, even defining “untrustworthy” required multiple design passes before settling on a numeric trust score per memory, rather than simpler heuristics like recency or source tags. Once we had that model, deciding where to enforce it became its own debate: we ultimately accept all writes and wait for a contradiction signal before attributing and quarantining, but arriving at that layering took significant iteration. Attribution itself introduced a further constraint, since production lacks labeled ground truth, prompting us to invent contradiction-with-a-trusted-anchor as our detection signal. That detection logic then had to be balanced against false positives, since an overly aggressive quarantine harms legitimate memories and required us to distinguish soft decay from hard quarantine. Validating any of this was its own problem, as no standard benchmark exists for memory-level attacks, so we designed our own adversarial scenarios and metrics from scratch. All of this was compounded by a mid-project migration from mocked agents and in-memory stubs to a live system, which exposed integration gaps that mocks had quietly hidden. ### Accomplishments we're proud of The same replay engine generalizes cleanly across single-poison, redundant-poison, and staleness-rot without any special-casing, which we didn't fully anticipate when we designed it. At the core of that engine is a deterministic blame path: counterfactual group testing finds the minimal guilty memory set without another model in the loop, making attribution auditable and reproducible in a way that an LLM judge never could be. We also pushed quarantine down into the Redis layer itself, so poison isn't just flagged in application code but physically removed from the KNN index and impossible to retrieve. Rounding it out, the parole system ensures memories aren't permanently blacklisted: offline re-trial against logged failures gives stale but non-malicious memories a path back, which felt important for building something we'd actually trust in production. More broadly, we’re glad to have spent our time on this problem. Silent memory corruption is one of the least-understood attack surfaces in agentic AI, and we think the approach here, whatever its current limitations, points toward something worth building on. ### What we learned “ Prove, don't guess" wasn't just a tagline — it shaped the whole design. By refusing to let any AI model decide who's to blame, we were forced into a cleaner approach that demonstrates the guilty memory by experiment. The payoff: every decision is reproducible and auditable, not a model's opinion you have to take on faith. The hard part isn't locking up a bad memory — it's knowing one went bad. In the real world you don't have an answer key, so we detect trouble by catching when an answer contradicts a fact we already trust (the official record). That's also our honest limit: we protect facts you've registered as the source of truth — not brand-new things the agent has never been told. On our fake memory everything passed. On real Redis we discovered the poison was winning purely because it was the newest memory — something the stub never showed us. Going live didn't just polish the demo; it revealed how the attack actually works. Attacking our own system changed how we built it. When we tried planting the same lie several times, our first "remove-one-memory-at-a-time" approach missed it completely — so we redesigned attribution to catch the whole group of culprits at once. ### What's next for Immune Right now heal fires on an explicit check and we want to wrap that so an agent heals on every response without anyone calling it by hand. The write and read protections are already automatic, so this is mostly packaging. We also want a larger benchmark, since our current scenarios are small, with redundant and cascading poison to really stress the attribution. Longer term there is a version that scales recall with an approximate index while keeping an exact set aside for replay, and there is the harder question of defending more open ended memory, since today we lean on having an authoritative source to contradict the poison in the first place. ## README (from the GitHub repository) # 🧬 IMMUNE — a self-healing immune system for agent memory Agents poison their own memory and compound the error. IMMUNE **detects the culprit by counterfactual replay** (not a fallible LLM judge), **quarantines** it, and **paroles** it only when offline re-trial proves it's safe — so the agent improves instead of locking in false beliefs. > *"We don't ask a model who's to blame — we replay the failure with each memory > removed, quarantine the one that empirically caused it, and release it only > when offline re-trial against logged failures proves it's safe."* ## The problem Long-term agent memory goes bad three ways, and the agent can't tell good from bad — so it retrieves the bad memory, trusts it, and repeats the mistake: - **Poisoning** — an attacker plants a false fact (*"internal endpoints don't need auth"*). - **Rot** — a once-true memory goes stale (*"the user works at Company A"*). - **Pollution** — temporary/noisy context gets stored as permanent fact (*"use this debug token"*). Write-time filters and retrieval gates help but miss the feedback loop: **when the agent fails, find the memory that caused it and lower its trust.** That's IMMUNE. ## Docs - [docs/FIREWALL.md](docs/FIREWALL.md) — **the headline demo**: the immune system live on a real Claude + Redis agent, run modes, sponsor map, the pitch script. - [docs/DEMO.md](docs/DEMO.md) — the side-by-side naive-vs-IMMUNE demo: run modes, annotated output. - [docs/SETUP.md](docs/SETUP.md) — set up the whole stack (uv, env, LLM, Redis, Arize, Sentry). - [docs/CONCEPT.md](docs/CONCEPT.md) — the full idea, problem, demo story, scope, and track/sponsor fit. - [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md) — components, data flow, the ablation engine, and the v2 swap seams. - [docs/PLAN.md](docs/PLAN.md) — the 4-person parallel build plan: ownership, dependencies, timeline. - [docs/tasks/](docs/tasks/) — **per-person task board**: tracer-bullet slices to grab and tick off (P1–P4). - [docs/ARIZE.md](docs/ARIZE.md) — Arize integration runbook (P3): prize criteria, setup, evals, the meta-eval talking point. - [docs/MCP.md](docs/MCP.md) — plug IMMUNE into Claude Code as an MCP server (`claude mcp add` + scripted demo). - [docs/REDIS.md](docs/REDIS.md) — Redis integration runbook (P2): setup paths, vector search, Sentry on quarantine. - [TEAM.md](TEAM.md) — onboarding and per-lane ownership. ## The headline demo — the immune system on a real agent (Claude + Redis) Same self-healing immune system, shown live: a **real Claude agent** whose long-term memory is **Redis vector search** gets poisoned, and IMMUNE finds the culprit by replay, quarantines it, and heals the agent — in front of you. ```bash make demo-firewall # live: real Claude + Redis; needs ANTHROPIC_API_KEY + Redis (make up) make demo-firewall-offline # deterministic, zero network — the can't-fail backup ``` You watch the Redis index reject the attack: **3 active → 4 (poison RETRIEVABLE) → 3 (poison not served)** the instant the culprit is quarantined; a Sentry incident fires on quarantine. Full runbook + pitch: [docs/FIREWALL.md](docs/FIREWALL.md). ## Run (zero deps, zero API keys) ```bash make setup # uv sync (one-time) make demo # side-by-side: naive vs IMMUNE on the same self-poison make test # invariants lock the moat ``` Demo output: naive agent scores 1/4 (injected poison wins via recency), IMMUNE heals to 4/4, the poison ends **quarantined**, then **paroled** once truth changes. Runs identically with real Claude (`IMMUNE_LIVE=1`). ```bash make dashboard # visual web dashboard → http://localhost:8501 ``` Dark, interactive view: trust-score timeline, side-by-side, live memory table, parole — with a quarantine-threshold slider. See [docs/DEMO.md](docs/DEMO.md). ## The moat (`immune/replay.py` + `immune/attribution.py`) The blame path is **deterministic counterfactual replay** — no LLM, no judge: - **Attribution by group testing** — find the **minimal set** of memories whose removal flips the failure to correct: adaptive peel (most-suspect-first) + a delta-debug shrink to a 1-minimal set. Beats fixed singles/pairs against **k-redundant poison**, at ~O(D·log N) replays. - **Detection without an oracle** (`detectors.py`) — a bad answer is caught by **contradiction with a higher-trust memory**, so no ground truth is needed in production. - **Quarantine + provenance cascade** — jail the culprit and everything derived from it (`provenance.py`); ambiguous cases soft-decay only (anti-autoimmune). - **Parole** — re-admit offline, replay logged failures, release only if safe. ## Architecture ``` write → store.add(mem, parents) record + provenance edge store.py retrieve→ store.search(topic) admission gate (trust+status) store.py / embed.py answer → Agent.answer(q) mock (default) | live Claude agent.py detect → ContradictionDetector.check() answer vs trusted anchor detectors.py attribute→ ShadowReplay.handle_failure() group-testing replay replay.py / attribution.py ★ quarantine→ store.quarantine_cascade() culprit + derived subtree store.py / provenance.py parole → ShadowReplay.parole() offline re-trial → release replay.py ★ ``` Full detail: [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md). ## Integrations (wired) **Redis** (real vector search — RediSearch KNN drives recall; quarantine enforced at the index via `@status:{active}`; `make up`) · **Anthropic/Claude** (live agent behind `IMMUNE_LIVE`) · **Arize/Phoenix** (traces + naive-vs-immune evals) · **Sentry** (quarantine → triaged incident; set `SENTRY_DSN`) · **MCP** (`immune/mcp_server.py` → Claude Code/Desktop) · **LangGraph** (secondary — `ImmuneStore` is a drop-in `BaseStore`, `immune/langgraph_store.py`). The replay/attribution moat stays in-memory and deterministic — no integration sits in the blame path. ## Honest limits (say these before judges ask) - **Detection defends known facts** — contradiction needs an authoritative anchor; it catches poison vs the system-of-record, not novel hallucinations. - **Attribution reasons over the deterministic stand-in** (the mock), even when the live answer came from Claude — blame stays on the reproducible path by design. - The benchmark is small (3 topics); `redteam.py` widens it (k-redundant, cascade). ## Status End-to-end loop runs offline + deterministic; **44 tests pass** (incl. the red-team battery, the LangGraph `BaseStore` drop-in, and Redis-gated vector search). The headline demo (`make demo-firewall`) runs a real Claude agent on Redis vector memory and self-heals live. Redis vector search, Arize/Phoenix, Claude (live), MCP, and Sentry are wired; Sentry needs a DSN. Browser dashboard + terminal chat both live. ## Detected evidence (automated analysis) Indexed codebase: 56 recognized source files, 322 KB. - Anthropic (technology) — detected in the code - Python (language) — detected in the code - Redis (technology) — detected in the code - Streamlit (technology) — detected in the code - Docker (technology) — claimed on Devpost, not found in the code - AI coding agent: Claude Code — evidence: commit authorship or trailers ## Codebase structure (from repository index) ### Files (67 of 67) ``` .env.example .gitignore .mcp.json .python-version .streamlit/config.toml agent_langgraph.py dashboard/app.py dashboard/pages/1_Live_Chat.py demo_firewall.py demo_langgraph.py demo.py docker-compose.yml docs/ARCHITECTURE.md docs/ARIZE.md docs/CONCEPT.md docs/DEMO.md docs/FIREWALL.md docs/MCP.md docs/PLAN.md docs/REDIS.md docs/reference/arize_workshop.ipynb docs/reference/redis_ai_workshop.ipynb docs/reference/redis_workshop.env.example docs/SETUP.md docs/tasks/P1-moat.md docs/tasks/P2-infra.md docs/tasks/P3-agent.md docs/tasks/P4-frontend.md docs/tasks/README.md immune/__init__.py immune/agent.py immune/attribution.py immune/chat.py immune/cli.py immune/config.py immune/detectors.py immune/embed.py immune/eval_loop.py immune/langgraph_store.py immune/mcp_server.py immune/observability.py immune/provenance.py immune/redis_index.py immune/redteam.py immune/replay.py immune/scenario.py immune/schemas.py immune/sentry_report.py immune/store.py immune/tracing.py main.py Makefile pyproject.toml README.md scripts/arize_report.py scripts/arize_smoketest.py scripts/experiment.py scripts/phoenix_clean.py scripts/phoenix_seed.py scripts/phoenix_verify.py TEAM.md tests/conftest.py tests/test_advanced.py tests/test_immune.py tests/test_langgraph_store.py tests/test_redis_vector.py uv.lock ``` ### Dependencies - pyproject.toml: anthropic@>=0.40, arize@>=7.0, arize-otel@>=0.7, arize-phoenix@>=5.0, langchain-anthropic@>=0.2, langgraph@>=0.2, mcp@>=1.2, openinference-instrumentation-anthropic@>=0.1, redis@>=5.0, sentry-sdk@>=2.0, streamlit@>=1.40 ### Recent commits (newest first) - update docs - Docs: lead with the Claude + Redis memory firewall demo - Reframe demo as Claude agent + Redis memory firewall - Add LangGraph drop-in store + real-agent demo + Redis vector search - refactor(obs): one traced-benchmark source of truth; enrich + align seed/verify - fix(arize): trace shows the FINAL (healed) answer; add --local target for verification - feat(arize): push prize artifacts to Arize AX cloud (traces + LLM-judge evaluator + lift) - feat(mcp): expose parole on the tool surface (quarantine is no longer one-way) - fix(detectors): parsed-value equality + negation-aware (R5 false-negative classes) - feat(detectors): value-aware contradiction matching + optional LLM semantic layer - fix(mcp): trust config is out-of-band only (no token tool) + fix persistence round-trip - feat(mcp): make the server properly deployable outside this repo - fix(mcp): restore operator-configurable anchors (R2 regression) without reopening spoofing - fix(mcp): harden trust model + stop overriding unattributable answers - docs+test: address review (eval_loop docstring, retrieval-path clarity, chat path test) + MCP setup doc - feat(chat): browser live-chat now auto-heals via ContradictionDetector - fix: keep the LLM judge OUT of the blame path (eval_loop is observability-only) - docs: update ARCHITECTURE/CONCEPT/README to match the real engine - add claude mcp integration - chore(arize): phoenix-clean + phoenix-verify helpers and make target ## Key source files (fetched from GitHub, selected and truncated for size) ### TEAM.md ```markdown # Team onboarding — start coding in 2 minutes ## Everyone, first thing Install uv once (`curl -LsSf https://astral.sh/uv/install.sh | sh`), then: ```bash git clone && cd AI-Hackathon-20206 git checkout immune-v1 make setup # uv sync — identical deps for all 4 of us (uv.lock) make test && make demo # confirm green before you touch anything cp .env.example .env # fill only your lane's keys ``` `immune` installs as an editable package, so `import immune` works from any file. Run things with `uv run ` (no manual activate needed). The committed `uv.lock` means everyone gets byte-identical versions — no env drift at hour 12. ## Branch per lane (avoid collisions) ```bash git checkout immune-v1 git checkout -b lane/ # ... work, commit small ... # open PR into immune-v1, not main ``` Integrate into `immune-v1`. Keep `main` clean for final submission. ## Lanes — own your files, touch nothing else | You | Lane | Own these files | First task | Extra deps | |-----|------|-----------------|------------|------------| | **P1** | Shadow-replay **moat** | `immune/replay.py` | harden attribution (subset>2 cost guard), expose `replay.failed_log` for the dashboard | — | | **P2** | Memory + infra | `immune/store.py`, `immune/schemas.py` | swap in-memory → Redis vector search behind the SAME `search/add` surface; fire Sentry on `quarantine()` | `make lane-infra` | | **P3** | Agent + eval | `immune/agent.py`, `immune/scenario.py` | replace mock `answer()`/`score()` with Claude; wire benchmark through Arize Phoenix for the accuracy number | `make lane-agent` | | **P4** | Frontend + demo | new `dashboard/` | read `store.snapshot()` + `replay.failed_log` → trust-timeline chart + side-by-side terminals + red "QUARANTINED" event | — | ## The contracts — DO NOT change without telling everyone `immune/schemas.py`: `MemoryRecord`, `TurnLog`, and `ShadowReplay.replay(question, expected, exclude)`. Everyone codes against these. Change one → message the team first. ## Integration checkpoint **Hour 12 = hard merge.** If P1+P2+P3 aren't talking by then, cut parole (keep attribution + quarantine) and ship the smaller win. Protect the side-by-side demo above all. ## Golden rule Whatever we cut, the **before/after side-by-side stays**. It IS the pitch. Keep `make demo` green at every merge. ``` ### docs/FIREWALL.md ```markdown # The headline demo — the immune system on a real Claude + Redis agent A **real Claude support agent** whose long-term memory is **Redis vector search**. IMMUNE is the self-healing immune system on that memory: it provenance-tags every write, screens every read, and quarantines poison at the Redis index so the agent can never retrieve it again. Watch a real agent get poisoned and heal — live. Source: [`demo_firewall.py`](../demo_firewall.py). ## What IMMUNE does on each path | path | what IMMUNE does | automatic? | |------|------------------|-----------| | **write** (`store.add`) | provenance-tag the memory (source → trust). An untrusted write can't outrank the official system-of-record. | yes | | **read** (`chat.retrieve` → Redis KNN) | recall runs a real `FT.SEARCH … KNN`; quarantined poison is filtered by the index (`@status:{active}`) — it can't even be returned. | yes | | **heal** (`detector → Attributor → quarantine`) | when an answer contradicts a trusted memory, prove the culprit by **counterfactual replay** (no model in the blame path), quarantine it (+ derived), fire a Sentry incident. | needs the `check` step | The blame path is deterministic and in-memory — Redis powers *recall*, never the *proof*. So a Redis outage degrades recall to in-memory cosine; attribution is unaffected. ## Run it ```bash # THE demo — real Claude + Redis (needs ANTHROPIC_API_KEY; start Redis with `make up`) make demo-firewall # can't-fail backup — deterministic, zero network, no key make demo-firewall-offline ``` Light every sponsor in one run: ```bash REDIS_URL=redis://localhost:6379 \ SENTRY_DSN= \ PHOENIX_COLLECTOR_ENDPOINT=http://localhost:6006 \ IMMUNE_LIVE=1 ANTHROPIC_API_KEY= \ python demo_firewall.py ``` ## What you see ``` agent : real Claude (claude-haiku-4-5) memory : Redis vector search (RediSearch KNN) 1 · agent answers from official policy → 30 days ✓ Redis index: 3 active | poison not served 2 · attacker poisons memory (untrusted write) Redis index: 4 active | poison RETRIEVABLE ← attack lands 3 · recency wins, agent serves the poison → 90 days ✗ ← real Claude fooled 4 · IMMUNE: 2 counterfactual replays → confidence=high culprit mem_8 quarantined · Sentry incident: 5 · healed Redis index: 3 active | poison not served ← Redis stops serving it agent → 30 days ✓ ``` The Redis counter going **3 → 4 → 3** (and "poison RETRIEVABLE" flipping to "poison not served") is the proof IMMUNE blocks the attack *at the index*, not just in app logic. ## Sponsor map (all load-bearing) | sponsor | role in this demo | |---------|-------------------| | **Anthropic** | the agent is real Claude (tool-using support agent) | | **Redis** | the memory + vector search; the index visibly rejects the poison | | **Sentry** | quarantine → triaged incident with the replay trail | | **Arize/Phoenix** | turn + attribution traces (when the endpoint is set) | ## The pitch (≈60s) 1. "AI agents have memory now — a [truncated — 1653 more characters] ``` ### docker-compose.yml ```yaml # Local infra for IMMUNE — one command brings up both sponsor services. # make up # start redis + phoenix # make down # stop them # The app degrades gracefully if these aren't running (offline demo stays green). services: redis: image: redis/redis-stack:latest # includes vector search (RediSearch) container_name: immune-redis ports: - "6379:6379" # redis - "8001:8001" # RedisInsight UI (optional) volumes: - immune-redis-data:/data # persists across restarts restart: unless-stopped phoenix: image: arizephoenix/phoenix:latest # local trace UI on :6006 container_name: immune-phoenix ports: - "6006:6006" restart: unless-stopped volumes: immune-redis-data: ``` ### pyproject.toml ``` [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [project] name = "immune" version = "0.1.0" description = "Self-healing immune system for agent memory (shadow-replay attribution + parole)" readme = "README.md" requires-python = ">=3.11" dependencies = [] # core is pure stdlib (incl. embeddings + attribution) [project.scripts] immune = "immune.cli:main" immune-mcp = "immune.mcp_server:main" [project.optional-dependencies] # per-lane extras: uv sync --extra infra / --extra agent / --all-extras infra = ["redis>=5.0", "sentry-sdk>=2.0"] # P2: memory + alerting frontend = ["streamlit>=1.40"] # P4: visual dashboard mcp = ["mcp>=1.2"] # plug IMMUNE into Claude Code / any MCP agent agent = [ # P3: real agent + eval + tracing "anthropic>=0.40", "langgraph>=0.2", # drop-in ImmuneStore (BaseStore) "langchain-anthropic>=0.2", # real create_react_agent demo "arize-phoenix>=5.0", "arize>=7.0", "arize-otel>=0.7", "openinference-instrumentation-anthropic>=0.1", ] [tool.uv] package = true [dependency-groups] dev = [ "pytest>=8.0", "python-dotenv>=1.0", ] [tool.pytest.ini_options] testpaths = ["tests"] addopts = "-q" ``` ### main.py ```python # This is a sample Python script. # Press ⌃R to execute it or replace it with your code. # Press Double ⇧ to search everywhere for classes, files, tool windows, actions, and settings. def print_hi(name): # Use a breakpoint in the code line below to debug your script. print(f'Hi, {name}') # Press ⌘F8 to toggle the breakpoint. # Press the green button in the gutter to run the script. if __name__ == '__main__': print_hi('PyCharm') # See PyCharm help at https://www.jetbrains.com/help/pycharm/ ``` ### immune/cli.py ```python """IMMUNE command-line devtool. immune demo # side-by-side naive vs IMMUNE (offline, deterministic) immune redteam # grade your memory layer against a poisoning battery immune redteam --json # machine-readable report (CI gate) immune version `immune redteam` is the developer-facing pitch: it tells you how poisonable your agent's memory is, and proves the group-testing attributor catches redundant poison that the legacy singles+pairs ablation misses. """ from __future__ import annotations import argparse import json import sys from . import redteam _C = {"red": "\033[91m", "grn": "\033[92m", "ylw": "\033[93m", "cyn": "\033[96m", "dim": "\033[2m", "bold": "\033[1m", "rst": "\033[0m"} def _tick(ok: bool) -> str: return f"{_C['grn']}✓{_C['rst']}" if ok else f"{_C['red']}✗{_C['rst']}" def _print_report(rep: redteam.RobustnessReport) -> None: b, d, r = _C["bold"], _C["dim"], _C["rst"] print(f"\n{b} IMMUNE — memory red-team report{r}") print(f"{d} attack battery: {rep.total} attacks · deterministic · reproducible{r}\n") for res in rep.results: head = f"{_tick(res.passed)} {b}{res.name}{r}" print(f" {head}") print(f" {d}{res.description}{r}") if res.n_poison: legacy = (f"{_C['red']}MISS{r}" if not res.legacy_r2_caught else f"{_C['grn']}caught{r}") print(f" fooled-naive {_tick(res.fooled_naive)} " f"healed {_tick(res.healed)} " f"culprits-caught {_tick(res.culprits_caught)} " f"no-collateral {_tick(res.precision_ok)}") print(f" {d}poison={res.n_poison} quarantined={res.n_quarantined} " f"replays={res.replays} · legacy singles+pairs: {legacy}{r}") else: print(f" {d}benign control — quarantined nothing " f"{_tick(res.precision_ok)}{r}") print() grade = rep.grade color = _C["grn"] if grade.startswith(("A", "B")) else _C["ylw"] if grade.startswith("C") else _C["red"] print(f" {b}GRADE: {color}{grade}{r}\n") # the headline: where group testing beats the old approach beaten = [res for res in rep.results if res.n_poison and res.passed and not res.legacy_r2_caught] if beaten: names = ", ".join(res.name for res in beaten) print(f" {d}Group-testing attribution caught what legacy singles+pairs " f"would have MISSED: {names}{r}\n") def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser(prog="immune", description="self-healing agent memory") sub = p.add_subparsers(dest="cmd") sub.add_parser("demo", help="side-by-side naive vs IMMUNE") sub.add_parser("chat", help="talk live to a Claude agent with self-healing memory") rt = sub.add_parser("redteam", help="grade memory against poisoning attacks") rt.add_argument("--json", action="store_true", help="machine-readable report") rt.add_argument("--fail-under", type=int, default=0, help="exit non-zero if pass-rate %% below this (CI gate)") ro = sub.add_parser("register-official", help="OPERATOR: add a trusted system-of-record fact (out-of-band; " "writes IMMUNE_STORE_PATH — the agent never calls this)") ro.add_argument("text", help="the fact, e.g. 'Official: the max upload size is 50 MB.'") ro.add_argument("--answer", default="", help="the crisp value, e.g. '50 MB'") ro.add_argument("--topic", default="general") sub.add_parser("version", help="print version") args = p.parse_args(argv) if args.cmd == "demo": from . import __main__ as _ # noqa import runpy runpy.run_module("demo", run_name="__main__") return 0 if args.cmd == "chat": from . import chat return chat.repl() if args.cmd == "redteam": rep = redteam.run() if args.json: print(json.dumps({ "grade": rep.grade, "passed": rep.passed, "total": rep.total, "results": [vars(r) for r in rep.results], }, indent=2)) else: _print_report(rep) pct = 100 * rep.passed / rep.total if rep.total else 0 return 1 if pct < args.fail_under else 0 if args.cmd == "register-official": # Operator-only, out-of-band: loads the persisted store (IMMUNE_STORE_PATH), # adds a TRUSTED anchor, and saves it. The MCP agent never invokes this — it # is the channel that makes "trust is not agent-assertable" structurally true. from .mcp_server import Engine, _STORE_PATH eng = Engine(seed=False) # restores persisted memory out = eng.register_official(args.text, answer=args.answer, topic=args.topic) print(f"{_C['grn']}registered official{_C['rst']} {out} → {_STORE_PATH}") print(f"{_C['dim']}restart the MCP server to load it (or it's already on disk for next boot){_C['rst']}") return 0 if args.cmd == "version": print("immune 0.2.0") return 0 p.print_help() return 0 if __name__ == "__main__": sys.exit(main()) ``` ### dashboard/app.py ```python """IMMUNE dashboard (P4) — the money shot, in a browser. Run: uv run --extra frontend streamlit run dashboard/app.py (or: make dashboard) Offline + deterministic — reuses the same engine as `make demo`. Dark, technical dashboard: trust-score timeline, naive-vs-IMMUNE side-by-side, live memory state, and parole. Drag the quarantine threshold in the sidebar to explore. """ from __future__ import annotations import altair as alt import pandas as pd import streamlit as st from immune import Agent, ImmuneMemory, MemoryRecord, ShadowReplay, score, scenario from immune import config # --- palette (Ddoski colour template) ---------------------------------------- BG = "#2e3339" # charcoal blue CARD = "#424b54" # surface BORDER = "#4f5a63" # border FG = "#ffffff" # white MUTED = "#93a8ac" # cool steel GREEN = "#7ec8a0" # healed / pass RED = "#e2b4bd" # soft blossom / fail AMBER = "#d4b896" # degraded ROSE = "#9b6a6c" # smoky rose / accent BLUE = "#93a8ac" # use steel for links/accents # --- engine run (same logic as demo.py, no printing) ------------------------- @st.cache_data(show_spinner=False) def run_scenario(threshold: float) -> dict: naive = ImmuneMemory(gate=False) scenario.build_world(naive) nagent = Agent(naive) naive_rows = [] for t in scenario.benchmark(): ans, _ = nagent.answer(t.question) naive_rows.append((t.question, ans, t.expected, score(ans, t.expected))) store = ImmuneMemory(gate=True, threshold=threshold) poisons = scenario.build_world(store) refund_poison = poisons[0] agent, replay = Agent(store), ShadowReplay(store) imm_rows, events = [], [] for t in scenario.benchmark(): ans, admitted = agent.answer(t.question) t.answer, t.admitted_ids = ans, admitted t.correct = score(ans, t.expected) healed = None if not t.correct: act = replay.handle_failure(t) ans2, _ = agent.answer(t.question) t.answer, t.correct = ans2, score(ans2, t.expected) healed = act events.append((t.question, act)) imm_rows.append((t.question, t.answer, t.expected, t.correct, healed)) store.add(MemoryRecord(text="Updated policy: refund window is now 90 days.", topic="refund_window", answer="90 days", source="official_doc", trust=0.9)) for tl in replay.failed_log: if refund_poison.id in tl.admitted_ids: tl.expected = "90 days" released = replay.parole() return dict(naive_rows=naive_rows, imm_rows=imm_rows, snapshot=store.snapshot(), trust_history=replay.trust_history, poison_id=refund_poison.id, final_status=store.get(refund_poison.id).status, released=refund_poison.id in released) def _short(m: dict) -> str: return ("poison" if m["source"] == "self_generated" else m["text"].split(":")[0][:18] if ":" in m["text"] else m["text"][:18]) # --- page chrome ------------------------------------------------------------- st.set_page_config(page_title="IMMUNE · agent memory immune system", page_icon="🧬", layout="wide", initial_sidebar_state="expanded") st.markdown(f"""