Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
142 changes: 142 additions & 0 deletions Gradata/src/gradata/_install_manifest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,142 @@
"""Install manifest for ``gradata install --agent <host>``.

Records, per host, the config file we wrote into and the SHA256 of that
file *immediately after* the install. ``gradata uninstall --agent <host>``
reads this manifest and:

- If the config file's current SHA matches the recorded SHA, the user
hasn't touched the file since install — safe to remove our entry.
- If the SHA differs, the user has edited the file by hand. Skip with a
``skipped X — modified since install`` message and leave the file
alone. (Better to leak one harmless hook entry than to clobber a
user-customized config.)

The manifest lives at ``~/.gradata/install_manifest.json`` so it survives
between ``install`` and ``uninstall`` invocations and is independent of
any single brain directory.

Schema (v1)::

{
"schema_version": 1,
"agents": {
"claude-code": {
"config_path": "/home/olive/.claude/settings.json",
"signature": "gradata:claude-code:/home/olive/brain",
"sha256_after_install": "abcdef…"
},
...
}
}
"""

from __future__ import annotations

import hashlib
import json
import os
from dataclasses import dataclass
from pathlib import Path

from gradata._atomic import atomic_write_text

SCHEMA_VERSION = 1


def manifest_path(home: Path | None = None) -> Path:
"""Return the on-disk manifest path. Honors $GRADATA_HOME if set."""
override = os.environ.get("GRADATA_HOME")
if override:
return Path(override) / "install_manifest.json"
resolved_home = home or Path(
os.environ.get("HOME") or os.environ.get("USERPROFILE") or os.path.expanduser("~")
)
return resolved_home / ".gradata" / "install_manifest.json"


@dataclass(frozen=True)
class AgentRecord:
config_path: Path
signature: str
sha256_after_install: str


def file_sha256(path: Path) -> str:
"""Return SHA256 of *path*'s bytes, or '' if it doesn't exist."""
if not path.exists():
return ""
h = hashlib.sha256()
with path.open("rb") as fh:
for chunk in iter(lambda: fh.read(65536), b""):
h.update(chunk)
return h.hexdigest()


def load(path: Path | None = None) -> dict:
"""Load the manifest as a plain dict. Returns empty structure if missing."""
p = path or manifest_path()
if not p.exists():
return {"schema_version": SCHEMA_VERSION, "agents": {}}
try:
data = json.loads(p.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
# Corrupt manifest — start fresh rather than crash uninstall.
return {"schema_version": SCHEMA_VERSION, "agents": {}}
if not isinstance(data, dict):
return {"schema_version": SCHEMA_VERSION, "agents": {}}
data.setdefault("schema_version", SCHEMA_VERSION)
data.setdefault("agents", {})
return data
Comment on lines +85 to +89
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Harden manifest shape validation before reading records.

Line 88 can leave agents as a non-dict (from malformed but valid JSON), and Line 121 then assumes .get() exists on it. That can crash uninstall/install flows instead of safely degrading.

🔧 Proposed fix
 def load(path: Path | None = None) -> dict:
@@
     if not isinstance(data, dict):
         return {"schema_version": SCHEMA_VERSION, "agents": {}}
     data.setdefault("schema_version", SCHEMA_VERSION)
-    data.setdefault("agents", {})
+    agents = data.get("agents")
+    if not isinstance(agents, dict):
+        data["agents"] = {}
     return data
@@
 def get_record(agent: str, path: Path | None = None) -> AgentRecord | None:
     data = load(path)
-    raw = data.get("agents", {}).get(agent)
+    agents = data.get("agents")
+    if not isinstance(agents, dict):
+        return None
+    raw = agents.get(agent)
@@
-    except KeyError:
+    except (KeyError, TypeError):
         return None

Also applies to: 119-131

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@Gradata/src/gradata/_install_manifest.py` around lines 85 - 89, The manifest
loader currently trusts that data["agents"] is a dict and can crash if JSON had
a non-dict value; update the validation logic around SCHEMA_VERSION and data so
that after confirming data is a dict you also enforce data["agents"] is a dict
(replace non-dict with {}), e.g., use isinstance checks on data and
data.get("agents") and call data.setdefault("agents", {}) only when agents is a
dict or otherwise assign an empty dict; apply the same hardening to the later
manifest-read/merge area (the block that references data.get(...) /
agents.get(...)) so all code paths treat agents as a dict.



def save(data: dict, path: Path | None = None) -> None:
p = path or manifest_path()
atomic_write_text(p, json.dumps(data, indent=2, sort_keys=True) + "\n")


def record_install(
agent: str,
config_path: Path,
signature: str,
*,
path: Path | None = None,
) -> None:
"""Record that *agent* installed a hook with *signature* into *config_path*.

Computes SHA256 of the config file after install and stores it. If a
record already exists for this agent we overwrite it (re-installs win).
"""
data = load(path)
agents = data.setdefault("agents", {})
agents[agent] = {
"config_path": str(config_path),
"signature": signature,
"sha256_after_install": file_sha256(config_path),
}
save(data, path)


def get_record(agent: str, path: Path | None = None) -> AgentRecord | None:
data = load(path)
raw = data.get("agents", {}).get(agent)
if not raw or not isinstance(raw, dict):
return None
try:
return AgentRecord(
config_path=Path(raw["config_path"]),
signature=raw["signature"],
sha256_after_install=raw.get("sha256_after_install", ""),
)
except KeyError:
return None


def drop_record(agent: str, path: Path | None = None) -> bool:
"""Remove *agent*'s record. Returns True if it existed."""
data = load(path)
agents = data.setdefault("agents", {})
if agent in agents:
del agents[agent]
save(data, path)
return True
return False
120 changes: 114 additions & 6 deletions Gradata/src/gradata/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -156,7 +156,7 @@ def cmd_status(args):
import time as _time
import urllib.error as _urllib_error
import urllib.request as _urllib_request
from datetime import datetime, timezone
from datetime import datetime

brain = _get_brain(args)
stats = brain.stats()
Expand Down Expand Up @@ -493,6 +493,19 @@ def _cmd_install_agent(args) -> None:
had_failure = True
print(f"{marker} {result.agent} → {result.config_path} ({result.action})")

# Record the install in ~/.gradata/install_manifest.json so that
# `gradata uninstall --agent <host>` can safely reverse it later
# (and detect user edits via SHA mismatch).
if result.action in ("added", "already_present"):
try:
from gradata._install_manifest import record_install
from gradata.hooks.adapters._base import hook_signature as _sig

record_install(name, config_path, _sig(name, brain_dir))
except Exception as exc:
# Manifest write is best-effort; don't fail install on it.
print(f" ⚠ install manifest write failed: {exc}")

# ▸ Flag-gated install verification: write + read a test rule
if verify_install and result.action != "failed":
try:
Expand Down Expand Up @@ -527,6 +540,82 @@ def _cmd_install_agent(args) -> None:
sys.exit(1)


def cmd_uninstall(args) -> None:
"""Reverse a prior ``gradata install --agent <host>``.

For each requested host:

- Look up the install manifest record (recorded by ``cmd_install`` at
install time). If absent, fall back to the canonical config path so
best-effort uninstall still works on installs that pre-date the
manifest.
- Compute the current SHA256 of the config file. If it differs from
the SHA recorded at install time, print
``skipped <host> — modified since install`` and leave the file
alone (preserve user edits).
- Otherwise call the adapter's ``uninstall(brain_dir, config_path)``
to symmetrically remove the entry the matching ``install()`` wrote.
- Always idempotent — running twice doesn't error.

Unknown hosts are rejected by argparse via the ``--agent`` choices
list (matching how ``cmd_install`` handles unknown hosts).
"""
from gradata._install_manifest import drop_record, file_sha256, get_record
from gradata.hooks.adapters._base import AGENTS, adapter_config_path, get_adapter

agent = args.agent
brain_dir = _resolve_brain_root(args)
agents = list(AGENTS) if agent == "all" else [agent]

had_failure = False
for name in agents:
try:
adapter = get_adapter(name)
except ValueError as exc:
# Shouldn't reach here because argparse choices guard this,
# but be defensive in case the CLI is invoked programmatically.
print(f"✗ {name} → unknown agent ({exc})")
had_failure = True
continue

record = get_record(name)
config_path = record.config_path if record else adapter_config_path(name)

# User-edit guard: skip uninstall if the config file's checksum
# differs from what was recorded at install time. Only meaningful
# when we have a recorded SHA — fall through for legacy installs.
if record and record.sha256_after_install:
current = file_sha256(config_path)
if current and current != record.sha256_after_install:
print(f"skipped {name} — modified since install")
continue

try:
result = adapter.uninstall(brain_dir, config_path)
except Exception as exc:
print(f"✗ {name} → unknown (failed: {exc})")
had_failure = True
continue

marker = "✓" if result.action != "failed" else "✗"
if result.action == "failed":
had_failure = True
print(f"{marker} {result.agent} → {result.config_path} ({result.action})")

# Drop manifest record only when we actually removed an entry
# (action == "added" in our adapter contract — "already_present"
# means nothing was there to remove, so we leave the manifest
# record alone in case the user re-installs).
if result.action == "added":
try:
Comment on lines +605 to +610
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Use the uninstall action when dropping manifest records.

Line 609 checks for "added", but uninstall adapters signal successful removal as "removed". This leaves stale manifest records after successful uninstall.

🔧 Proposed fix
-        # Drop manifest record only when we actually removed an entry
-        # (action == "added" in our adapter contract — "already_present"
-        # means nothing was there to remove, so we leave the manifest
-        # record alone in case the user re-installs).
-        if result.action == "added":
+        # Drop manifest record only when uninstall actually removed an entry.
+        if result.action == "removed":
             try:
                 drop_record(name)
             except Exception as exc:
                 print(f"  ⚠ install manifest update failed: {exc}")
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@Gradata/src/gradata/cli.py` around lines 605 - 610, The current uninstall
logic in the block that decides whether to drop the manifest record incorrectly
checks if result.action == "added"; change the check to look for the uninstall
success action ("removed") so manifest records are deleted only when the adapter
signaled removal. Locate the conditional that reads result.action == "added" in
the uninstall/remove flow in Gradata/src/gradata/cli.py and replace it to test
for "removed" (leaving handling for "already_present" untouched).

drop_record(name)
except Exception as exc:
print(f" ⚠ install manifest update failed: {exc}")

if had_failure:
sys.exit(1)


def cmd_recall(args) -> None:
from gradata.mcp_tools import gradata_recall

Expand Down Expand Up @@ -783,20 +872,20 @@ def cmd_prove(args):
den = sum((x - mean_x) ** 2 for x in xs) or 1.0
slope = num / den

print(f"Corrections per session:")
print("Corrections per session:")
print(f" Sessions: {n}")
print(f" Total corrections: {sum(counts)}")
print(f" Mean: {mean_y:.1f}/session")
if n >= 3:
print(f" Trend slope: {slope:+.3f} corrections/session")
if slope < -0.05:
print(f" Verdict: CONVERGING (brain is learning — fewer corrections over time)")
print(" Verdict: CONVERGING (brain is learning — fewer corrections over time)")
elif slope > 0.05:
print(f" Verdict: DIVERGING (corrections rising — brain may need tuning)")
print(" Verdict: DIVERGING (corrections rising — brain may need tuning)")
else:
print(f" Verdict: STABLE (flat trend)")
print(" Verdict: STABLE (flat trend)")
else:
print(f" Trend: need >=3 sessions to estimate")
print(" Trend: need >=3 sessions to estimate")

# Rule application rate
total_apps = sum(rule_apps_by_session.values())
Expand Down Expand Up @@ -1955,6 +2044,24 @@ def main():
help="Port for the daemon when used with --systemd (default: 8765)",
)

# uninstall — symmetrical reverse of `install --agent <host>` (GRA-1241)
p_uninstall = sub.add_parser(
"uninstall",
help="Reverse `gradata install --agent <host>` — remove agent hook/MCP config",
)
p_uninstall.add_argument(
"--agent",
required=True,
choices=["claude-code", "codex", "gemini", "cursor", "hermes", "opencode", "all"],
help="Agent whose hook/MCP config to uninstall",
)
p_uninstall.add_argument(
"--brain",
type=str,
default=None,
help="Brain directory the hook points at (default: BRAIN_DIR or ./brain)",
)

# health
p_health = sub.add_parser("health", help="Brain health report")
p_health.add_argument("--json", action="store_true")
Expand Down Expand Up @@ -2214,6 +2321,7 @@ def main():
"validate": cmd_validate,
"doctor": cmd_doctor,
"install": cmd_install,
"uninstall": cmd_uninstall,
"health": cmd_health,
"report": cmd_report,
"watch": cmd_watch,
Expand Down
57 changes: 56 additions & 1 deletion Gradata/src/gradata/hooks/adapters/_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@

from gradata._atomic import atomic_write_text

Action = Literal["added", "already_present", "failed"]
Action = Literal["added", "already_present", "failed", "removed"]

AGENTS = ("claude-code", "codex", "gemini", "cursor", "hermes", "opencode")
_MODULES = {
Expand Down Expand Up @@ -165,6 +165,61 @@ def failure(agent: str, config_path: Path, exc: Exception) -> InstallResult:
return InstallResult(agent, config_path, "failed", str(exc))


def uninstall_from_list_in_dict(
*,
agent: str,
brain_dir: Path,
agent_config_path: Path,
outer_key: str,
inner_key: str,
) -> InstallResult:
"""Generic uninstall helper for JSON configs shaped as ``{outer:{inner:[entries]}}``.

Used by gemini (tools.preCall), opencode (hooks.preTool), hermes
(hooks.pre_tool_call). Removes every entry whose ``str(entry)``
contains our ``hook_signature(agent, brain_dir)``. Idempotent.
Empty containers are pruned.
"""
try:
if not agent_config_path.is_file():
return InstallResult(
agent, agent_config_path, "already_present", "config file does not exist"
)
sig = hook_signature(agent, brain_dir)
data = read_json(agent_config_path)
outer = data.get(outer_key)
if not isinstance(outer, dict):
return InstallResult(
agent, agent_config_path, "already_present", f"no {outer_key} block"
)
entries = outer.get(inner_key)
if not isinstance(entries, list):
return InstallResult(
agent, agent_config_path, "already_present", f"no {outer_key}.{inner_key}"
)

removed = 0
kept: list = []
for entry in entries:
if sig in str(entry):
removed += 1
continue
kept.append(entry)
if removed == 0:
return InstallResult(agent, agent_config_path, "already_present", "hook not present")

if kept:
outer[inner_key] = kept
else:
outer.pop(inner_key, None)
if not outer:
data.pop(outer_key, None)
write_json(agent_config_path, data)
return InstallResult(agent, agent_config_path, "removed", f"removed {removed} hook entry")
except Exception as exc:
return failure(agent, agent_config_path, exc)


# --------------------------------------------------------------------------
# Shared extraction utilities — used by individual adapters.
# --------------------------------------------------------------------------
Expand Down
Loading
Loading