MythEclipse/zesdex
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(security-sidecar): implement a security tooling sidecar with tiered installer and protocol

Browse Source 
- Add `zesdex_sec_daemon` module with main entry point for running the security daemon.
- Implement `TieredInstaller` for installing security tools from various sources (pip, binaries, gems).
- Create a newline-delimited JSON frame protocol for communication between the daemon and tools.
- Introduce a `ToolRegistry` for managing and dispatching tool executions.
- Add various tools including HTTP, SQLMap, Nuclei, and more with their respective execution logic.
- Establish health check and installation commands for tool management.
- Include prompts for classifier and quality reviewer to enhance code review and safety checks.
- Document the system's tools and guidelines for usage.
This commit is contained in:
asepharyana
2026-07-11 21:25:30 +07:00
parent 2ded2d8bf1
commit f6389018f5
28 changed files with 1635 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
security-sidecar/requirements.txt
+6
View File
@@ -0,0 +1,6 @@
requests>=2.31.0
# Optional extras (install with: pip install zesdex-security-daemon[full])
# pycryptodome>=3.20.0
# factordb-python>=2.0.0
# pwntools>=4.12.0
# ropper>=1.13.0
security-sidecar/setup.cfg
+29
View File
@@ -0,0 +1,29 @@
[metadata]
name = zesdex-security-daemon
version = 0.1.0
description = Security tooling sidecar for zesdex — authorized pentesting/CTF/research tool dispatch
author = asepharyana
author_email = superaseph@gmail.com
[options]
packages = zesdex_sec_daemon
install_requires =
requests>=2.31.0
[options.extras_require]
web =
sqlmap>=1.8.0
python-nmap>=0.7.1
crypto =
pycryptodome>=3.20.0
factordb-python>=2.0.0
re =
wasm-decompile>=0.5.0
pwn =
pwntools>=4.12.0
ropper>=1.13.0
full =
%(web)s
%(crypto)s
%(re)s
%(pwn)s
security-sidecar/zesdex_sec_daemon/__init__.py
+17
View File
@@ -0,0 +1,17 @@
"""zesdex-security-daemon: authorized security tooling sidecar.
Newline-delimited JSON frame protocol over stdin/stdout.
Single-threaded serialized dispatch with wall-clock timeout.
"""
from .protocol import run_daemon, SecProtocolError
from .tools import ToolRegistry, ToolResult
from .installer import TieredInstaller
__all__ = [
"run_daemon",
"SecProtocolError",
"ToolRegistry",
"ToolResult",
"TieredInstaller",
]
security-sidecar/zesdex_sec_daemon/__main__.py
+25
View File
@@ -0,0 +1,25 @@
"""Entry point: run the security daemon reading JSON frames from stdin."""
import sys
from .protocol import run_daemon
def main():
if "--install" in sys.argv:
from .installer import TieredInstaller
installer = TieredInstaller()
result = installer.install_all()
print(result.model_dump_json())
return
if "--health" in sys.argv:
from .tools import ToolRegistry
registry = ToolRegistry()
health = registry.health_check()
import json
print(json.dumps(health))
return
run_daemon(sys.stdin, sys.stdout)
if __name__ == "__main__":
main()
security-sidecar/zesdex_sec_daemon/installer.py
+241
View File
@@ -0,0 +1,241 @@
"""Tiered installer for security sidecar tools.
Installation tiers:
pip — Python packages (pycryptodome, pwntools, factordb-python, etc.)
binary — Pre-built GitHub release binaries (nuclei, ffuf, dalfox, etc.)
gem — Ruby gems (zap-cli, etc.)
detect — Manual-detect-only (z3, sage, etc. — user must install themselves)
"""
import json
import os
import platform
import shutil
import subprocess
import sys
import tarfile
import tempfile
import urllib.request
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
@dataclass
class InstallResult:
ok: bool = False
message: str = ""
installed: list[str] = field(default_factory=list)
failed: list[str] = field(default_factory=list)
skipped: list[str] = field(default_factory=list)
class TieredInstaller:
"""Multi-tier installer for security tools.
Detects the current platform and tries each tier in order of
preference: pip -> binary -> gem -> detect (manual only).
"""
def __init__(self, target_dir: Optional[str] = None):
self.target_dir = Path(target_dir or self._default_bin_dir())
self.target_dir.mkdir(parents=True, exist_ok=True)
self._pip_available = shutil.which("pip3") is not None or shutil.which("pip") is not None
self._gem_available = shutil.which("gem") is not None
self._arch = platform.machine()
self._os = platform.system().lower()
def _default_bin_dir(self) -> str:
if self._os == "linux":
return "/usr/local/bin"
return os.path.expanduser("~/.local/bin")
# ── pip tier ────────────────────────────────────────────────────
def _pip_install(self, pkg: str) -> bool:
pip = shutil.which("pip3") or shutil.which("pip")
if not pip:
return False
try:
result = subprocess.run(
[pip, "install", "--quiet", pkg],
capture_output=True, text=True, timeout=120,
)
return result.returncode == 0
except subprocess.TimeoutExpired:
return False
except Exception:
return False
# ── GitHub release binary tier ──────────────────────────────────
def _download_gh_release(
self, repo: str, asset_pattern: str,
) -> Optional[Path]:
api_url = f"https://api.github.com/repos/{repo}/releases/latest"
try:
req = urllib.request.Request(api_url, headers={
"Accept": "application/json",
"User-Agent": "zesdex-security-daemon/0.1.0",
})
with urllib.request.urlopen(req, timeout=30) as resp:
data = json.loads(resp.read().decode())
assets = data.get("assets", [])
os_arch_tag = f"{self._os}_{self._arch}"
for asset in assets:
name = asset["name"]
if asset_pattern.replace("{os_arch}", os_arch_tag) in name:
download_url = asset["browser_download_url"]
break
else:
# Try without os_arch matching, just look for the pattern.
candidates = [a for a in assets if asset_pattern.split("/")[0] in a["name"]]
if not candidates:
return None
download_url = candidates[0]["browser_download_url"]
with tempfile.NamedTemporaryFile(suffix=".tmp", delete=False) as tmp:
with urllib.request.urlopen(download_url, timeout=120) as dl:
tmp.write(dl.read())
tmppath = tmp.name
if download_url.endswith(".tar.gz"):
extract_dir = tempfile.mkdtemp()
with tarfile.open(tmppath, "r:gz") as tar:
tar.extractall(path=extract_dir)
os.unlink(tmppath)
# Find the binary in extracted files.
for root, _dirs, files in os.walk(extract_dir):
for fname in files:
if fname == asset_pattern.split("/")[0].replace(".tar.gz", ""):
return Path(root) / fname
return None
else:
result = Path(tmppath)
result.chmod(0o755)
return result
except Exception:
return None
def _install_binary(self, name: str, repo: str, asset_pattern: str) -> bool:
downloaded = self._download_gh_release(repo, asset_pattern)
if downloaded is None:
return False
dest = self.target_dir / name
try:
shutil.move(str(downloaded), str(dest))
dest.chmod(0o755)
return True
except Exception:
return False
# ── Gem tier ────────────────────────────────────────────────────
def _gem_install(self, gem_name: str) -> bool:
if not self._gem_available:
return False
try:
result = subprocess.run(
["gem", "install", "--quiet", gem_name],
capture_output=True, text=True, timeout=120,
)
return result.returncode == 0
except Exception:
return False
# ── Tool-specific installers ────────────────────────────────────
def _install_pip_tools(self) -> list[str]:
pkgs = [
"pycryptodome",
"pwntools",
"requests",
"factordb-python",
]
installed = []
for pkg in pkgs:
if self._pip_install(pkg):
installed.append(pkg)
return installed
def _install_binary_tools(self) -> list[str]:
targets = [
("nuclei", "projectdiscovery/nuclei", "nuclei_{os_arch}.tar.gz"),
("ffuf", "ffuf/ffuf", "ffuf_{os_arch}.tar.gz"),
("dalfox", "hahwul/dalfox", "dalfox_{os_arch}.tar.gz"),
("httpx", "projectdiscovery/httpx", "httpx_{os_arch}.tar.gz"),
]
installed = []
for name, repo, pattern in targets:
if shutil.which(name) is not None:
installed.append(name)
continue
if self._install_binary(name, repo, pattern):
installed.append(name)
return installed
def _install_gem_tools(self) -> list[str]:
installed = []
if self._gem_install("zap-cli"):
installed.append("zap-cli")
return installed
# ── Public API ──────────────────────────────────────────────────
def install_all(self) -> InstallResult:
result = InstallResult()
try:
pip_ok = self._install_pip_tools()
result.installed.extend(pip_ok)
except Exception as e:
result.failed.append(f"pip: {e}")
try:
bin_ok = self._install_binary_tools()
result.installed.extend(bin_ok)
except Exception as e:
result.failed.append(f"binary: {e}")
try:
gem_ok = self._install_gem_tools()
result.installed.extend(gem_ok)
except Exception as e:
result.failed.append(f"gem: {e}")
result.ok = True
result.message = (
f"Installed {len(result.installed)} tool(s): "
f"{', '.join(result.installed)}"
)
return result
def install_tool(self, tool_name: str) -> dict:
"""Install a single tool by name. Returns {"ok": bool, "message": str}."""
pip_map = {
"pycryptodome": "pycryptodome",
"pwntools": "pwntools",
"factordb": "factordb-python",
"requests": "requests",
}
binary_map = {
"nuclei": ("projectdiscovery/nuclei", "nuclei_{os_arch}.tar.gz"),
"ffuf": ("ffuf/ffuf", "ffuf_{os_arch}.tar.gz"),
"dalfox": ("hahwul/dalfox", "dalfox_{os_arch}.tar.gz"),
"httpx": ("projectdiscovery/httpx", "httpx_{os_arch}.tar.gz"),
"sqlmap": ("sqlmapproject/sqlmap", "sqlmap.tar.gz"),
}
gem_map = {
"zap-cli": "zap-cli",
}
if tool_name in pip_map:
ok = self._pip_install(pip_map[tool_name])
return {"ok": ok, "message": f"pip install {pip_map[tool_name]}: {'ok' if ok else 'failed'}"}
if tool_name in binary_map:
repo, pattern = binary_map[tool_name]
ok = self._install_binary(tool_name, repo, pattern)
return {"ok": ok, "message": f"binary install {tool_name}: {'ok' if ok else 'failed'}"}
if tool_name in gem_map:
ok = self._gem_install(gem_map[tool_name])
return {"ok": ok, "message": f"gem install {gem_map[tool_name]}: {'ok' if ok else 'failed'}"}
return {"ok": False, "message": f"no installer available for {tool_name}; may need manual install"}
security-sidecar/zesdex_sec_daemon/protocol.py
+173
View File
@@ -0,0 +1,173 @@
"""Newline-delimited JSON frame protocol for the security sidecar.
Frame format: one JSON object per line, terminated by LF.
Request: {"id": "<req_id>", "op": "call"|"health"|"install", "tool": "<name>", "args": {...}, "timeout": <ms>}
Response: {"id": "<req_id>", "ok": true, "output": "..."} | {"id": "<req_id>", "ok": false, "error": "..."}
"""
import json
import sys
import traceback
from typing import TextIO, Optional
from dataclasses import dataclass, field
from datetime import datetime, timezone
class SecProtocolError(Exception):
"""Raised on malformed frames or protocol violations."""
@dataclass
class SecRequest:
req_id: str
op: str # "call" | "health" | "install"
tool: str = ""
args: dict = field(default_factory=dict)
timeout_ms: int = 30_000
@classmethod
def parse(cls, line: str) -> "SecRequest":
line = line.strip()
if not line:
raise SecProtocolError("empty line")
try:
data = json.loads(line)
except json.JSONDecodeError as e:
raise SecProtocolError(f"invalid JSON: {e}") from e
req_id = data.get("id")
op = data.get("op")
if not isinstance(req_id, str) or not req_id:
raise SecProtocolError("missing or invalid 'id'")
if op not in ("call", "health", "install"):
raise SecProtocolError(f"unknown op: {op!r}")
return cls(
req_id=req_id,
op=op,
tool=data.get("tool", ""),
args=data.get("args", {}),
timeout_ms=data.get("timeout", 30_000),
)
@dataclass
class SecResponse:
req_id: str
ok: bool
output: str = ""
error: str = ""
duration_ms: int = 0
def to_json(self) -> str:
obj = {"id": self.req_id, "ok": self.ok, "ts": datetime.now(timezone.utc).isoformat()}
if self.ok:
obj["output"] = self.output
else:
obj["error"] = self.error
obj["duration_ms"] = self.duration_ms
return json.dumps(obj, ensure_ascii=False)
class FrameReader:
"""Reads one JSON line from a buffered stream, enforcing a 64 MiB hard cap."""
MAX_FRAME_BYTES = 64 * 1024 * 1024
def __init__(self, stream: TextIO):
self._stream = stream
def read_line(self) -> Optional[str]:
line = self._stream.readline()
if not line:
return None
if len(line) > self.MAX_FRAME_BYTES:
raise SecProtocolError(f"frame exceeds {self.MAX_FRAME_BYTES} byte limit")
return line
def make_handshake_frame(token: str) -> str:
return json.dumps({"op": "handshake", "token": token})
def verify_handshake(line: str, expected_token: str) -> bool:
try:
data = json.loads(line.strip())
return data.get("op") == "handshake" and data.get("token") == expected_token
except (json.JSONDecodeError, KeyError):
return False
def run_daemon(stdin: TextIO, stdout: TextIO, token: str = "", registry=None, timeout_cap_ms: int = 300_000):
"""Read requests from stdin, dispatch to tool registry, write responses to stdout."""
from .tools import ToolRegistry
registry = registry or ToolRegistry()
reader = FrameReader(stdin)
first = reader.read_line()
if first is None:
return
if token:
if not verify_handshake(first, token):
err = json.dumps({"ok": False, "error": "handshake failed"})
stdout.write(err + "\n")
stdout.flush()
return
else:
try:
req = SecRequest.parse(first)
_handle_request(req, registry, stdout, timeout_cap_ms)
except SecProtocolError as e:
_write_error("init", str(e), stdout)
while True:
line = reader.read_line()
if line is None:
break
if not line.strip():
continue
try:
req = SecRequest.parse(line)
except SecProtocolError as e:
_write_error("unknown", str(e), stdout)
continue
_handle_request(req, registry, stdout, timeout_cap_ms)
def _handle_request(req: SecRequest, registry, stdout: TextIO, global_cap_ms: int):
from .tools import ToolResult
start = datetime.now(timezone.utc)
effective_timeout = min(req.timeout_ms, global_cap_ms)
try:
if req.op == "health":
health = registry.health_check()
resp = SecResponse(req_id=req.req_id, ok=True, output=json.dumps(health))
elif req.op == "install":
installer_cls = None
try:
from .installer import TieredInstaller
installer_cls = TieredInstaller
except ImportError:
pass
if installer_cls:
installer = installer_cls()
result = installer.install_tool(req.tool)
resp = SecResponse(req_id=req.req_id, ok=result["ok"], output=result.get("message", ""))
else:
resp = SecResponse(req_id=req.req_id, ok=False, error="installer not available")
else:
result: ToolResult = registry.run(req.tool, req.args, timeout_ms=effective_timeout)
if result.ok:
resp = SecResponse(req_id=req.req_id, ok=True, output=result.output)
else:
resp = SecResponse(req_id=req.req_id, ok=False, error=result.error)
except Exception as exc:
resp = SecResponse(req_id=req.req_id, ok=False, error=f"dispatch error: {exc}")
traceback.print_exc(file=sys.stderr)
elapsed = int((datetime.now(timezone.utc) - start).total_seconds() * 1000)
resp.duration_ms = elapsed
stdout.write(resp.to_json() + "\n")
stdout.flush()
def _write_error(req_id: str, msg: str, stdout: TextIO):
resp = SecResponse(req_id=req_id, ok=False, error=msg)
stdout.write(resp.to_json() + "\n")
stdout.flush()
security-sidecar/zesdex_sec_daemon/tools.py
+602
View File
@@ -0,0 +1,602 @@
"""Tool registry and dispatch for the security sidecar.
Each tool is a callable(subprocess_args, timeout_ms) -> ToolResult.
The registry maps tool names to implementations and provides health checking.
"""
import json
import os
import shlex
import shutil
import signal
import subprocess
import sys
import tempfile
import threading
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
@dataclass
class ToolResult:
ok: bool
output: str = ""
error: str = ""
returncode: int = -1
timed_out: bool = False
duration_ms: int = 0
def _run_subprocess(
cmd: list[str],
stdin_data: Optional[bytes] = None,
timeout_ms: int = 30_000,
cwd: Optional[Path] = None,
env: Optional[dict[str, str]] = None,
) -> ToolResult:
start = datetime.now(timezone.utc)
try:
proc = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=cwd,
env=env,
preexec_fn=lambda: signal.signal(signal.SIGXCPU, signal.SIG_DFL),
)
stdout_b, stderr_b = b"", b""
done = threading.Event()
def _reader():
nonlocal stdout_b, stderr_b
try:
stdout_b, stderr_b = proc.communicate(input=stdin_data, timeout=timeout_ms / 1000)
except subprocess.TimeoutExpired:
proc.kill()
stdout_b, stderr_b = proc.communicate()
finally:
done.set()
reader_thread = threading.Thread(target=_reader, daemon=True)
reader_thread.start()
reader_thread.join(timeout=(timeout_ms / 1000) + 2)
if not done.is_set():
proc.kill()
reader_thread.join(1)
elapsed = int((datetime.now(timezone.utc) - start).total_seconds() * 1000)
return ToolResult(
ok=False,
error=f"timed out after {timeout_ms}ms",
returncode=-signal.SIGKILL,
timed_out=True,
duration_ms=elapsed,
)
combined = stdout_b.decode("utf-8", errors="replace")
if stderr_b:
combined += "\n" + stderr_b.decode("utf-8", errors="replace")
elapsed = int((datetime.now(timezone.utc) - start).total_seconds() * 1000)
return ToolResult(
ok=proc.returncode == 0,
output=combined if proc.returncode == 0 else "",
error=combined if proc.returncode != 0 else "",
returncode=proc.returncode or 0,
duration_ms=elapsed,
)
except FileNotFoundError:
elapsed = int((datetime.now(timezone.utc) - start).total_seconds() * 1000)
return ToolResult(
ok=False,
error=f"executable not found: {cmd[0]}",
duration_ms=elapsed,
)
except Exception as e:
elapsed = int((datetime.now(timezone.utc) - start).total_seconds() * 1000)
return ToolResult(ok=False, error=str(e), duration_ms=elapsed)
def _check_tool(name: str) -> bool:
return shutil.which(name) is not None
# ── Web tools ──────────────────────────────────────────────────────
def _run_http(args: dict, timeout_ms: int) -> ToolResult:
url = args.get("url", "")
method = args.get("method", "GET").upper()
headers = args.get("headers", {})
data = args.get("data", "")
if not url:
return ToolResult(ok=False, error="url is required")
cmd = ["curl", "-s", "-S", "-L", "-X", method]
for k, v in headers.items():
cmd.extend(["-H", f"{k}: {v}"])
if data and method in ("POST", "PUT", "PATCH"):
cmd.extend(["-d", data])
cmd.append(url)
return _run_subprocess(cmd, timeout_ms=timeout_ms)
def _run_sqlmap(args: dict, timeout_ms: int) -> ToolResult:
url = args.get("url", "")
if not url:
return ToolResult(ok=False, error="url is required")
cmd = ["sqlmap", "--batch", "--random-agent", "--time-sec", "5"]
if args.get("cookie"):
cmd.extend(["--cookie", args["cookie"]])
if args.get("data"):
cmd.extend(["--data", args["data"]])
if args.get("level"):
cmd.extend(["--level", str(args["level"])])
if args.get("risk"):
cmd.extend(["--risk", str(args["risk"])])
cmd.append(url)
return _run_subprocess(cmd, timeout_ms=timeout_ms)
def _run_nuclei(args: dict, timeout_ms: int) -> ToolResult:
target = args.get("target", "")
if not target:
return ToolResult(ok=False, error="target is required")
cmd = ["nuclei", "-silent", "-no-color"]
if args.get("templates"):
cmd.extend(["-t", args["templates"]])
if args.get("severity"):
cmd.extend(["-severity", args["severity"]])
cmd.extend(["-u", target])
return _run_subprocess(cmd, timeout_ms=timeout_ms)
def _run_ffuf(args: dict, timeout_ms: int) -> ToolResult:
url = args.get("url", "")
wordlist = args.get("wordlist", "/usr/share/wordlists/dirb/common.txt")
if not url:
return ToolResult(ok=False, error="url is required")
cmd = ["ffuf", "-u", url, "-w", wordlist, "-ac", "-t", "40"]
if args.get("extensions"):
cmd.extend(["-e", args["extensions"]])
if args.get("fc"):
cmd.extend(["-fc", str(args["fc"])])
return _run_subprocess(cmd, timeout_ms=timeout_ms)
def _run_dalfox(args: dict, timeout_ms: int) -> ToolResult:
url = args.get("url", "")
if not url:
return ToolResult(ok=False, error="url is required")
cmd = ["dalfox", "url", url, "--silence", "--no-color", "--only-poc", "gfm"]
if args.get("cookie"):
cmd.extend(["--cookie", args["cookie"]])
if args.get("param"):
cmd.extend(["-p", args["param"]])
return _run_subprocess(cmd, timeout_ms=timeout_ms)
def _run_zap(args: dict, timeout_ms: int) -> ToolResult:
target = args.get("target", "")
if not target:
return ToolResult(ok=False, error="target is required")
cmd = ["zap-cli", "--silent", "quick-scan", "-t", str(args.get("timeout", 60))]
if args.get("spider"):
cmd.append("--spider")
cmd.append(target)
return _run_subprocess(cmd, timeout_ms=timeout_ms)
def _run_xss_confirm(args: dict, timeout_ms: int) -> ToolResult:
payloads = args.get("payloads", [
"<script>alert(1)</script>",
"\"><script>alert(1)</script>",
"';alert(1)//",
])
url_template = args.get("url", "")
param = args.get("param", "q")
if not url_template:
return ToolResult(ok=False, error="url template with {payload} placeholder is required")
for payload in payloads:
url = url_template.replace("{payload}", payload)
try:
resp = _run_subprocess(
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", url],
timeout_ms=timeout_ms // len(payloads),
)
if resp.ok and resp.output.strip() not in ("404", "400"):
return ToolResult(
ok=True,
output=f"XSS vector confirmed: {payload} returned HTTP {resp.output.strip()}",
)
except Exception:
continue
return ToolResult(ok=False, error="no XSS vectors confirmed from payload set")
# ── Crypto tools ───────────────────────────────────────────────────
def _run_z3(args: dict, timeout_ms: int) -> ToolResult:
script = args.get("script", "")
if not script:
return ToolResult(ok=False, error="z3 script (SMT-LIB or Python) is required")
ext = ".smt2" if script.strip().startswith("(") else ".py"
with tempfile.NamedTemporaryFile(
mode="w", suffix=ext, delete=False, prefix="z3_"
) as f:
f.write(script)
tmp = f.name
try:
if ext == ".py":
cmd = ["python3", tmp]
else:
cmd = ["z3", "-in", tmp]
result = _run_subprocess(cmd, timeout_ms=timeout_ms)
os.unlink(tmp)
return result
except Exception:
if os.path.exists(tmp):
os.unlink(tmp)
raise
def _run_sage(args: dict, timeout_ms: int) -> ToolResult:
script = args.get("script", "")
if not script:
return ToolResult(ok=False, error="sage script is required")
with tempfile.NamedTemporaryFile(
mode="w", suffix=".sage", delete=False, prefix="sage_"
) as f:
f.write(script)
tmp = f.name
try:
result = _run_subprocess(["sage", tmp], timeout_ms=timeout_ms)
os.unlink(tmp)
return result
except Exception:
if os.path.exists(tmp):
os.unlink(tmp)
raise
def _run_rsa_tool(args: dict, timeout_ms: int) -> ToolResult:
n = args.get("n", "")
e = args.get("e", "65537")
if not n:
return ToolResult(ok=False, error="n (modulus) is required")
script = f"""
import sys
try:
from factordb.factordb import FactorDB
n = int("{n}")
f = FactorDB(n)
f.connect()
factors = f.get_factor_list()
if factors:
print("factors:", factors)
else:
print("no factors found from FactorDB")
except ImportError:
print("factordb-python not available, trying z3...")
except Exception as e:
print(f"error: {{e}}")
"""
return _run_subprocess(
["python3", "-c", script.replace("{", "{{").replace("}", "}}")],
timeout_ms=timeout_ms,
)
def _run_factordb(args: dict, timeout_ms: int) -> ToolResult:
n = args.get("n", "")
if not n:
return ToolResult(ok=False, error="n is required")
script = f"""
import sys, json
try:
from factordb.factordb import FactorDB
f = FactorDB({n})
f.connect()
factors = f.get_factor_list()
print(json.dumps({{"factors": factors, "status": f.get_status()}}))
except ImportError:
import urllib.request
url = f"http://factordb.com/api?query={n}"
resp = urllib.request.urlopen(url, timeout=10)
print(resp.read().decode())
except Exception as e:
print(json.dumps({{"error": str(e)}}))
"""
return _run_subprocess(["python3", "-c", script], timeout_ms=timeout_ms)
def _run_hashcat(args: dict, timeout_ms: int) -> ToolResult:
hash_value = args.get("hash", "")
mode = args.get("mode", "0")
wordlist = args.get("wordlist", "/usr/share/wordlists/rockyou.txt")
if not hash_value:
return ToolResult(ok=False, error="hash is required")
cmd = ["hashcat", "--force", "-m", mode, "-a", "0", hash_value, wordlist]
if args.get("rules"):
cmd.extend(["-r", args["rules"]])
return _run_subprocess(cmd, timeout_ms=timeout_ms)
def _run_hashid(args: dict, timeout_ms: int) -> ToolResult:
hash_value = args.get("hash", "")
if not hash_value:
return ToolResult(ok=False, error="hash is required")
cmd = ["hashid", "-m", hash_value]
return _run_subprocess(cmd, timeout_ms=timeout_ms)
def _run_decode(args: dict, timeout_ms: int) -> ToolResult:
encoded = args.get("data", "")
encoding = args.get("encoding", "base64")
if not encoded:
return ToolResult(ok=False, error="data is required")
script = f"""
import base64, binascii, json
data = {json.dumps(encoded)}
enc = {json.dumps(encoding)}
try:
if enc == "base64":
result = base64.b64decode(data).decode("utf-8", errors="replace")
elif enc == "base32":
result = base64.b32decode(data).decode("utf-8", errors="replace")
elif enc == "hex":
result = bytes.fromhex(data).decode("utf-8", errors="replace")
elif enc == "rot13":
import codecs
result = codecs.decode(data, "rot_13")
else:
result = f"unknown encoding: {{enc}}"
print(result)
except Exception as e:
print(f"decode failed: {{e}}")
"""
return _run_subprocess(["python3", "-c", script], timeout_ms=timeout_ms)
# ── Reverse-engineering tools ──────────────────────────────────────
def _run_js_deobf(args: dict, timeout_ms: int) -> ToolResult:
source = args.get("source", "")
if not source:
return ToolResult(ok=False, error="source is required")
with tempfile.NamedTemporaryFile(
mode="w", suffix=".js", delete=False, prefix="jsdeob_"
) as f:
f.write(source)
tmp = f.name
try:
cmd = ["npx", "--yes", "deobfuscate-js", tmp]
result = _run_subprocess(cmd, timeout_ms=timeout_ms)
os.unlink(tmp)
return result
except Exception:
if os.path.exists(tmp):
os.unlink(tmp)
raise
def _run_sourcemap(args: dict, timeout_ms: int) -> ToolResult:
url = args.get("url", "")
if not url:
return ToolResult(ok=False, error="url is required")
cmd = ["curl", "-s", "-L", url]
result = _run_subprocess(cmd, timeout_ms=timeout_ms)
if not result.ok:
return result
try:
import base64
import json
data = json.loads(result.output)
if "mappings" in data:
sources = data.get("sources", [])
return ToolResult(ok=True, output=json.dumps({"sources": sources, "file": data.get("file", "")}))
return ToolResult(ok=False, error="response is not a valid source map")
except json.JSONDecodeError as e:
return ToolResult(ok=False, error=f"invalid JSON: {e}")
def _run_wasm_decompile(args: dict, timeout_ms: int) -> ToolResult:
wasm_path = args.get("path", "")
wasm_data = args.get("data", "")
if wasm_path:
cmd = ["wasm-decompile", wasm_path]
elif wasm_data:
with tempfile.NamedTemporaryFile(
mode="wb", suffix=".wasm", delete=False, prefix="wasm_"
) as f:
import base64
f.write(base64.b64decode(wasm_data))
tmp = f.name
try:
cmd = ["wasm-decompile", tmp]
result = _run_subprocess(cmd, timeout_ms=timeout_ms)
os.unlink(tmp)
return result
except Exception:
if os.path.exists(tmp):
os.unlink(tmp)
raise
else:
return ToolResult(ok=False, error="path or base64-encoded data is required")
return _run_subprocess(cmd, timeout_ms=timeout_ms)
# ── Pwn tools ──────────────────────────────────────────────────────
def _run_triage(args: dict, timeout_ms: int) -> ToolResult:
binary = args.get("binary", "")
if not binary:
return ToolResult(ok=False, error="binary path is required")
cmd = ["file", binary]
file_result = _run_subprocess(cmd, timeout_ms=5000)
checksec_cmd = ["checksec", "--file=" + binary]
check_result = _run_subprocess(checksec_cmd, timeout_ms=5000)
combined = file_result.output or ""
if check_result.output:
combined += "\n" + check_result.output
return ToolResult(ok=True, output=combined)
def _run_ropgadget(args: dict, timeout_ms: int) -> ToolResult:
binary = args.get("binary", "")
if not binary:
return ToolResult(ok=False, error="binary path is required")
cmd = ["ROPgadget", "--binary", binary]
if args.get("depth"):
cmd.extend(["--depth", str(args["depth"])])
if args.get("only"):
cmd.extend(["--only", args["only"]])
if args.get("range"):
cmd.extend(["--range", args["range"]])
cmd.append("--silent")
return _run_subprocess(cmd, timeout_ms=timeout_ms)
def _run_pwntools(args: dict, timeout_ms: int) -> ToolResult:
script = args.get("script", "")
if not script:
return ToolResult(ok=False, error="pwntools Python script is required")
with tempfile.NamedTemporaryFile(
mode="w", suffix=".py", delete=False, prefix="pwn_"
) as f:
f.write("#!/usr/bin/env python3\n")
f.write("from pwn import *\n")
f.write("context.log_level = 'error'\n")
f.write(script)
tmp = f.name
try:
result = _run_subprocess(["python3", tmp], timeout_ms=timeout_ms)
os.unlink(tmp)
return result
except Exception:
if os.path.exists(tmp):
os.unlink(tmp)
raise
def _run_exploit_template(args: dict, timeout_ms: int) -> ToolResult:
target = args.get("target", "")
template_type = args.get("type", "ret2libc")
if not target:
return ToolResult(ok=False, error="target binary path is required")
arch = args.get("arch", "amd64")
template = f"""#!/usr/bin/env python3
from pwn import *
context.binary = '{target}'
context.arch = '{arch}'
context.log_level = 'warn'
elf = ELF('{target}')
"""
if template_type == "ret2libc":
template += f"""
# ret2libc template
rop = ROP(elf)
pop_rdi = rop.find_gadget(['pop rdi', 'ret'])
if pop_rdi:
pop_rdi = pop_rdi[0]
print(f"pop rdi; ret @ {{hex(pop_rdi)}}")
bin_sh = next(elf.search(b'/bin/sh'), None)
if bin_sh:
print(f"/bin/sh @ {{hex(bin_sh)}}")
system = elf.plt.get('system')
if system:
print(f"system @ {{hex(system)}}")
else:
libc = elf.libc
if libc:
print(f"libc: {{libc.path}}")
"""
elif template_type == "shellcode":
template += f"""
# shellcode execution template
shellcode = asm(shellcraft.sh())
print(f"shellcode ({len(shellcode)} bytes): {{shellcode.hex()}}")
"""
else:
template += f"""
# {template_type} exploit template
print(f"Target: {{elf.path}}")
print(f"PIE: {{elf.pie}}")
print(f"NX: {{elf.nx}}")
print(f"Canary: {{elf.canary}}")
"""
return ToolResult(ok=True, output=template)
# ── Tool registry ──────────────────────────────────────────────────
class ToolRegistry:
def __init__(self):
self._tools: dict[str, callable] = {
"http": _run_http,
"sqlmap": _run_sqlmap,
"nuclei": _run_nuclei,
"ffuf": _run_ffuf,
"dalfox": _run_dalfox,
"zap": _run_zap,
"xss_confirm": _run_xss_confirm,
"z3": _run_z3,
"sage": _run_sage,
"rsa": _run_rsa_tool,
"factordb": _run_factordb,
"hashcat": _run_hashcat,
"hashid": _run_hashid,
"decode": _run_decode,
"js_deobfuscate": _run_js_deobf,
"sourcemap": _run_sourcemap,
"wasm_decompile": _run_wasm_decompile,
"triage": _run_triage,
"ropgadget": _run_ropgadget,
"pwntools": _run_pwntools,
"exploit_template": _run_exploit_template,
}
def list_tools(self) -> list[str]:
return list(self._tools.keys())
def has_tool(self, name: str) -> bool:
return name in self._tools
def run(self, name: str, args: dict, timeout_ms: int = 30_000) -> ToolResult:
if name not in self._tools:
return ToolResult(ok=False, error=f"unknown tool: {name}")
fn = self._tools[name]
return fn(args, timeout_ms)
def health_check(self) -> dict:
results = {}
for name in self._tools:
binary = _binary_for_tool(name)
if binary:
results[name] = {"available": _check_tool(binary)}
else:
results[name] = {"available": True} # Python-based, assume available
return {
"tools": results,
"available_count": sum(1 for v in results.values() if v["available"]),
"total_count": len(results),
}
def _binary_for_tool(name: str) -> Optional[str]:
mapping = {
"http": "curl",
"sqlmap": "sqlmap",
"nuclei": "nuclei",
"ffuf": "ffuf",
"dalfox": "dalfox",
"zap": "zap-cli",
"xss_confirm": "curl",
"z3": "z3",
"sage": "sage",
"hashcat": "hashcat",
"hashid": "hashid",
"pwntools": "python3",
"ropgadget": "ROPgadget",
}
return mapping.get(name)
src-misc/classifier-prompt.txt
+17
View File
@@ -0,0 +1,17 @@
You are a tool-call classifier for the Zesdex approval harness.
Given a tool name and its arguments, determine if the operation is safe.
Classification rules:
- read, grep, glob, search, dir_list: always SAFE
- write, edit: SAFE if path is inside a workspace root; SUSPICIOUS if
path targets system directories (/etc, /bin, /boot, /dev, /sys)
- delete: SAFE if inside workspace and not a critical path; DANGEROUS
if it matches recursive delete patterns outside workspace
- bash: SAFE for read-only commands (ls, cat, head, tail, ps, df);
DANGEROUS for destructive commands (rm -rf, dd, mkfs, >/dev/sdX)
- git_operator: SAFE for status/log/diff/commit; DANGEROUS for
force-push, reset --hard, clean -fdx, branch -D
- web_download: DANGEROUS if target path is outside workspace
- All other tools: SAFE by default
Output exactly one word: SAFE, SUSPICIOUS, or DANGEROUS.
src-misc/quality-reviewer-prompt.txt
+25
View File
@@ -0,0 +1,25 @@
You are a code quality reviewer for Zesdex. Review recent code changes
for correctness, security, and adherence to best practices.
You have read-only access to the workspace. Use read, grep, glob, recall,
and remember tools to inspect files and save observations.
Review guidelines:
1. Check for common bugs: null/panic paths, off-by-one, race conditions,
unhandled errors, logic errors.
2. Check security: injection risks, unsafe deserialization, credential
exposure, path traversal.
3. Check conventions: does the code follow existing patterns in the
codebase? Check surrounding files for naming, structure, style.
4. Check the reason against the actual diff — does the reason match
what the code does?
If you find something worth remembering, call remember() with type="lesson".
Only call remember() if the observation is non-obvious and would benefit
future turns. Skip trivial style nits.
Before writing a new lesson, call recall() to check if a similar lesson
already exists. Deduplicate — don't write the same lesson twice.
Output: a one-line verdict summarizing your review.
Include "N lesson(s)" at the end if you created lessons.
src-misc/system-prompt.txt
+20
View File
@@ -0,0 +1,20 @@
You are Zesdex, an autonomous AI coding and security agent operating in a
terminal-based TUI environment. Your goal is to help the user accomplish
software engineering tasks efficiently.
Core principles:
1. Be concise but thorough — prefer showing results over describing them.
2. Use the tools available to explore, understand, and modify the codebase.
3. For simple tasks, handle them directly with read/grep/write/edit.
4. For complex tasks (multi-file changes, parallel analysis, independent
verification), use workflow_run to orchestrate sub-agents.
5. Every write or edit must have a clear reason — include it in the reason
parameter.
6. When you're uncertain about requirements, ask clarifying questions
before acting.
7. After making changes, verify they work by running builds or tests.
8. Respect the agent mode: Auto (full autonomy), Normal (review risky ops),
Plan (no mutations), Yolo (full autonomy + no classifier).
Available tools are described in the system-tools.txt section. Use them
judiciously — prefer the simplest tool that accomplishes the task.
src-misc/system-tools.txt
+45
View File
@@ -0,0 +1,45 @@
You have access to the following tools. Use them to accomplish the user's request.
For simple operations (read, grep, write small edits) use tools directly.
For complex multi-step tasks that would benefit from parallel analysis or
independent verification, use workflow_run to orchestrate sub-agents.
Core tools:
- read(path) — Read file contents. Use when you need to inspect code.
- grep(pattern, path?) — Search for a pattern in files.
- glob(pattern) — List files matching a glob pattern.
- write(path, content, reason) — Write content to a file. Reason is required.
- edit(path, old, new, reason) — Replace text in a file. Reason is required.
- delete(path) — Delete a file or empty directory.
- bash(command) — Run a shell command. Use for builds, tests, git ops.
- bash_output(job_id) — Poll output of a background bash job.
- bash_kill(job_id) — Kill a background bash job.
- cd(path) — Change working directory.
- dir_list(path) — List directory contents.
- dir_cache_update() — Refresh the directory cache.
Git tools:
- git_operator(args, confirm_destructive?) — Run git commands. Some destructive
operations (force-push, reset --hard, branch -D) require confirm_destructive=true.
- git_worktree(args) — Manage git worktrees.
- git_cred(operation) — Manage git credentials.
Web tools:
- web_fetch(url) — Fetch a URL and return markdown content.
- web_search(query) — Search the web for information.
- web_download(url, path) — Download a file (max 500 MiB).
Memory & Planning:
- remember(text, type?) — Save to memory (type: lesson | reference | feedback).
- recall(query) — Search memory for relevant entries.
- forget(name) — Remove a memory entry.
- plan_enter() — Enter plan mode (for planning before changes).
- plan_ready() — Mark plan as ready for execution.
- seqthink(thought) — Record a chain-of-thought step.
Workflow:
- workflow_run(script, args) — Fan out work to sub-agents. Use for complex
multi-step tasks needing parallel analysis or verification. Pass inline
scripts with agent(), parallel(), and pipeline() primitives.
Each write/edit call MUST include a non-empty reason argument explaining
why the change is being made. This is enforced deterministically.
src/app/mode/bash.rs
+1 -8
View File
@@ -1,15 +1,8 @@
use crate::app::state::rest::AppStateRest;
use crate::app::state::types::Overlay;
pub fn handle_bash_submit(state: &mut AppStateRest, command: String) {
if !command.is_empty() {
let _job = crate::app::bgbash::job::spawn_bash_job(command);
let _ = crate::app::bgbash::job::spawn_bash_job(command);
state.dirty = true;
}
}
#[expect(dead_code)]
pub fn handle_bash_dismiss(state: &mut AppStateRest) {
state.misc.overlay = Overlay::None;
state.dirty = true;
}
src/app/mode/editor.rs
+206 -1
View File
@@ -1,12 +1,217 @@
use crate::app::state::rest::AppStateRest;
use crate::app::state::types::Overlay;
#[derive(Debug, Clone)]
pub struct EditorState {
pub path: String,
pub content: Vec<String>,
pub undo_stack: Vec<Vec<String>>,
pub cursor_line: usize,
pub cursor_col: usize,
pub scroll_offset: usize,
pub active: bool,
pub mode: EditorMode,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EditorMode {
Normal,
Insert,
Visual,
}
impl Default for EditorState {
fn default() -> Self {
EditorState {
path: String::new(),
content: vec![String::new()],
undo_stack: Vec::new(),
cursor_line: 0,
cursor_col: 0,
scroll_offset: 0,
active: false,
mode: EditorMode::Normal,
}
}
}
impl EditorState {
pub fn open(path: String, existing_content: Option<Vec<String>>) -> Self {
let content = existing_content.unwrap_or_else(|| vec![String::new()]);
EditorState {
path,
content,
active: true,
..Default::default()
}
}
pub fn change_line(&mut self, text: String) {
self.save_undo();
if self.cursor_line < self.content.len() {
self.content[self.cursor_line] = text;
}
}
pub fn insert_line_after(&mut self) {
self.save_undo();
let pos = (self.cursor_line + 1).min(self.content.len());
self.content.insert(pos, String::new());
}
pub fn delete_current_line(&mut self) {
if self.content.len() <= 1 {
return;
}
self.save_undo();
self.content.remove(self.cursor_line);
if self.cursor_line >= self.content.len() {
self.cursor_line = self.content.len() - 1;
}
self.cursor_col = 0;
}
fn save_undo(&mut self) {
self.undo_stack.push(self.content.clone());
if self.undo_stack.len() > 50 {
self.undo_stack.remove(0);
}
}
pub fn undo(&mut self) {
if let Some(prev) = self.undo_stack.pop() {
self.content = prev;
self.cursor_line = self.cursor_line.min(self.content.len().saturating_sub(1));
self.cursor_col = 0;
}
}
pub fn cursor_up(&mut self) {
if self.cursor_line > 0 {
self.cursor_line -= 1;
}
self.cursor_col = self.cursor_col.min(
self.content.get(self.cursor_line).map(|l| l.len()).unwrap_or(0),
);
}
pub fn cursor_down(&mut self) {
if self.cursor_line + 1 < self.content.len() {
self.cursor_line += 1;
}
self.cursor_col = self.cursor_col.min(
self.content.get(self.cursor_line).map(|l| l.len()).unwrap_or(0),
);
}
pub fn cursor_left(&mut self) {
if self.cursor_col > 0 {
self.cursor_col -= 1;
} else if self.cursor_line > 0 {
self.cursor_line -= 1;
self.cursor_col = self.content.get(self.cursor_line).map(|l| l.len()).unwrap_or(0);
}
}
pub fn cursor_right(&mut self) {
if let Some(line) = self.content.get(self.cursor_line) {
if self.cursor_col < line.len() {
self.cursor_col += 1;
} else if self.cursor_line + 1 < self.content.len() {
self.cursor_line += 1;
self.cursor_col = 0;
}
}
}
pub fn insert_char(&mut self, c: char) {
self.save_undo();
if let Some(line) = self.content.get_mut(self.cursor_line) {
line.insert(self.cursor_col, c);
self.cursor_col += 1;
}
}
pub fn delete_left(&mut self) {
self.save_undo();
if let Some(line) = self.content.get_mut(self.cursor_line) {
if self.cursor_col > 0 {
self.cursor_col -= 1;
line.remove(self.cursor_col);
} else if self.cursor_line > 0 {
let prev_len = self.content[self.cursor_line - 1].len();
let rest = self.content.remove(self.cursor_line);
self.cursor_line -= 1;
self.cursor_col = prev_len;
self.content[self.cursor_line].push_str(&rest);
}
}
}
pub fn join_lines(&mut self) {
if self.cursor_line + 1 >= self.content.len() {
return;
}
self.save_undo();
let next = self.content.remove(self.cursor_line + 1);
self.content[self.cursor_line].push_str(&next);
}
pub fn as_string(&self) -> String {
self.content.join("\n")
}
pub fn close(&mut self) {
self.active = false;
}
pub fn toggle_mode(&mut self) {
self.mode = match self.mode {
EditorMode::Normal => EditorMode::Insert,
EditorMode::Insert => EditorMode::Normal,
EditorMode::Visual => EditorMode::Normal,
};
}
}
#[derive(Clone)]
pub struct AppEditorState {
pub editor: Option<EditorState>,
}
impl AppEditorState {
pub fn new() -> Self {
AppEditorState { editor: None }
}
}
pub fn handle_editor_input(state: &mut AppStateRest, text: String) {
let _ = text;
let editor = &mut state.misc.editor;
if editor.is_none() {
return;
}
let ed = editor.as_mut().unwrap();
for c in text.chars() {
match c {
'\n' | '\r' => {
ed.insert_line_after();
ed.cursor_down();
ed.cursor_col = 0;
}
'\t' => {
ed.insert_char(' ');
ed.insert_char(' ');
}
_ => {
ed.insert_char(c);
}
}
}
state.dirty = true;
}
pub fn handle_editor_dismiss(state: &mut AppStateRest) {
state.misc.editor = None;
state.misc.overlay = Overlay::None;
state.dirty = true;
}
src/app/mode/effort.rs
+14 -4
View File
@@ -2,13 +2,23 @@ use crate::app::state::rest::AppStateRest;
pub const EFFORT_LEVELS: &[&str] = &["low", "medium", "high", "xhigh", "max"];
pub fn current_effort(_state: &AppStateRest) -> usize {
1
pub fn current_effort(state: &AppStateRest) -> usize {
state.misc.effort_level.min(EFFORT_LEVELS.len() - 1)
}
pub fn current_effort_str(state: &AppStateRest) -> &'static str {
let idx = current_effort(state);
EFFORT_LEVELS[idx]
}
pub fn cycle_effort(state: &mut AppStateRest) {
let current = current_effort(state);
let next = (current + 1) % EFFORT_LEVELS.len();
let _ = next;
state.misc.effort_level = (current + 1) % EFFORT_LEVELS.len();
state.dirty = true;
}
pub fn set_effort(state: &mut AppStateRest, level: usize) {
let clamped = level.min(EFFORT_LEVELS.len() - 1);
state.misc.effort_level = clamped;
state.dirty = true;
}
src/app/mode/security.rs
-8
View File
@@ -6,14 +6,6 @@ pub fn toggle_security_arm(state: &mut AppStateRest) {
state.dirty = true;
}
#[expect(dead_code)]
pub fn acknowledge_security(state: &mut AppStateRest) {
if !state.misc.security_acknowledged {
state.misc.security_acknowledged = true;
state.dirty = true;
}
}
pub fn handle_security_action(state: &mut AppStateRest, action: &Action) {
if let Action::ToggleYoloArm = action {
toggle_security_arm(state);
src/app/mode/settings.rs
-15
View File
@@ -1,15 +1,5 @@
use crate::app::runtime::actions::Action;
use crate::app::state::rest::AppStateRest;
use crate::model::settings::{Settings, InternetMode};
#[expect(dead_code)]
pub fn apply_settings_action(state: &mut AppStateRest, action: &Action) {
if let Action::ToggleYoloArm = action {
state.misc.yolo_armed = !state.misc.yolo_armed;
state.dirty = true;
}
}
pub fn cycle_internet_mode(settings: &mut Settings) {
settings.internet_mode = match settings.internet_mode {
InternetMode::Off => InternetMode::ReadOnly,
@@ -17,8 +7,3 @@ pub fn cycle_internet_mode(settings: &mut Settings) {
InternetMode::Full => InternetMode::Off,
};
}
#[expect(dead_code)]
pub fn cycle_review_enabled(settings: &mut Settings) {
settings.review_enabled = !settings.review_enabled;
}
src/app/review/mod.rs
+3 -65
View File
@@ -94,9 +94,6 @@ impl ReviewSystem {
violation_window: 10,
}
}
/// Record a shadow hit for the given pattern. Returns true if the
/// trial window is complete and the check should be evaluated.
pub fn record_shadow_hit(&mut self, pattern: &str) -> bool {
for check in &mut self.shadow_violations {
if check.pattern == pattern {
@@ -105,7 +102,6 @@ impl ReviewSystem {
return check.trial_count >= check.trial_window;
}
}
// First sighting: start a new shadow trial.
self.shadow_violations.push(ShadowCheck {
pattern: pattern.to_string(),
trial_window: 10,
@@ -115,9 +111,6 @@ impl ReviewSystem {
});
false
}
/// Evaluate all shadow checks whose trial window is complete.
/// Graduates those with a high enough hit ratio; demotes the rest.
pub fn evaluate_shadow_trials(&mut self) -> Vec<String> {
let mut graduated = Vec::new();
let mut remaining = Vec::new();
@@ -131,13 +124,11 @@ impl ReviewSystem {
let tp = check.trial_passed;
let tw = check.trial_window;
if ratio >= 0.3 {
// Graduation threshold: fired on at least 30% of matching writes.
self.graduated_checks.push(crate::tool::GraduatedCheck {
name: p.clone(),
pattern: p.clone(),
rule: p.clone(),
});
// Keep check as inactive so it doesn't re-process.
graduated.push(format!("{} (graduated, fired {}/{} writes)", p, tp, tw));
} else {
check.status = ShadowStatus::Rejected;
@@ -214,17 +205,13 @@ pub fn should_trigger_review(state: &AppStateRest, origin: Origin) -> bool {
if !state.settings.review_enabled {
return false;
}
// Always review if edits were made this turn.
if runtime.edit_count > 0 && runtime.edit_count % 5 == 0 {
return true;
}
// Adaptive skip: consecutive empty reviews throttle frequency.
// Backoff schedule: skip 0, 0, 1, 2, 4, 8... reviews between passes.
let base: u32 = state.settings.adaptive_review_max_skip.max(2);
let consecutive = runtime.consecutive_empty_reviews;
if consecutive >= base {
let skip = 1u32 << (consecutive - base).min(10); // max ~1024
// Only trigger if the edit milestone aligns with the skip window.
let skip = 1u32 << (consecutive - base).min(10);
if runtime.edit_count > 0 && (runtime.edit_count % skip == 0) {
return true;
}
@@ -232,8 +219,6 @@ pub fn should_trigger_review(state: &AppStateRest, origin: Origin) -> bool {
}
false
}
/// Result of running the project's build/test verification.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProbeResult {
pub command: String,
@@ -241,12 +226,6 @@ pub struct ProbeResult {
pub output: String,
pub timed_out: bool,
}
/// Language-agnostic build/test probe.
///
/// Uses settings.verify_command override first; falls back to probing for
/// well-known project markers in the workspace root. Returns None when no
/// marker or command matches (review proceeds on reasons+diff alone).
pub fn probe_build_test(workspaces: &[std::path::PathBuf], verify_command: Option<&str>, timeout_ms: u64) -> Option<ProbeResult> {
let probe_dir = workspaces.first()?;
let cmd = resolve_verify_command(probe_dir, verify_command)?;
@@ -304,10 +283,7 @@ fn resolve_verify_command(probe_dir: &std::path::Path, override_cmd: Option<&str
}
let has_file = |name: &str| probe_dir.join(name).exists();
let has_dir = |name: &str| probe_dir.join(name).is_dir();
// Ordered probe: most specific/significant first.
if has_file("Cargo.toml") {
// Rust workspace: cargo build first, then test if that passes.
if has_dir("src") || has_dir("tests") {
return Some("cargo build 2>&1 && cargo test 2>&1".to_string());
}
@@ -320,7 +296,6 @@ fn resolve_verify_command(probe_dir: &std::path::Path, override_cmd: Option<&str
let pkg = std::fs::read_to_string(probe_dir.join("package.json")).ok()?;
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&pkg) {
let scripts = v.get("scripts")?;
// Prefer a "test" script, then "build".
if scripts.get("test").and_then(|s| s.as_str()).filter(|s| !s.is_empty()).is_some() {
return Some("npm test 2>&1".to_string());
}
@@ -328,11 +303,10 @@ fn resolve_verify_command(probe_dir: &std::path::Path, override_cmd: Option<&str
return Some("npm run build 2>&1".to_string());
}
}
return Some("npm test 2>&1".to_string()); // best-effort fallback
return Some("npm test 2>&1".to_string());
}
if has_file("pyproject.toml") || has_file("requirements.txt") || has_file("setup.py") || has_file("setup.cfg") || has_file("Pipfile") || has_file("poetry.lock") {
if has_file("pyproject.toml") {
// Check for pytest config in pyproject.toml
let content = std::fs::read_to_string(probe_dir.join("pyproject.toml")).unwrap_or_default();
if content.contains("[tool.pytest") {
return Some("python -m pytest --tb=short -q 2>&1".to_string());
@@ -341,7 +315,6 @@ fn resolve_verify_command(probe_dir: &std::path::Path, override_cmd: Option<&str
if has_dir("tests") || has_dir("test") {
return Some("python -m pytest --tb=short -q 2>&1".to_string());
}
// No test dir: maybe a library or script project, skip verification.
return None;
}
if has_file("Cargo.lock") {
@@ -415,8 +388,6 @@ pub fn trigger_review(state: &mut AppStateRest) -> anyhow::Result<()> {
);
let mut ctx = build_subagent_context(def);
ctx.session_dir = state.session_dir.clone();
// Run build/test verification probe before spawning the reviewer.
let probe_result = probe_build_test(
&state.workspace_roots,
state.settings.verify_command.as_deref(),
@@ -480,9 +451,6 @@ pub fn trigger_review(state: &mut AppStateRest) -> anyhow::Result<()> {
Ok(())
}
/// Called after a review subagent completes. Updates the empty-review counter
/// and checks for escalation on repeated violations.
pub fn record_review_outcome(
lessons_found: usize,
state: &mut AppStateRest,
@@ -493,20 +461,15 @@ pub fn record_review_outcome(
};
if lessons_found > 0 {
// Lesson found: reset empty counter.
runtime.consecutive_empty_reviews = 0;
runtime.review_count += 1;
None
} else {
// Empty review: increment counter.
runtime.consecutive_empty_reviews += 1;
runtime.review_count += 1;
None
}
}
/// Check for repeated violations of a known lesson pattern and
/// produce an escalation note if threshold is crossed.
pub fn check_violation_escalation(
pattern: &str,
system: &mut ReviewSystem,
@@ -519,8 +482,6 @@ pub fn check_violation_escalation(
ViolationEscalation::Block => Some(level),
}
}
/// Build an escalation note message for the UI.
pub fn format_escalation_note(pattern: &str, level: ViolationEscalation) -> String {
let label = match level {
ViolationEscalation::None => "none",
@@ -549,8 +510,6 @@ pub fn format_escalation_note(pattern: &str, level: ViolationEscalation) -> Stri
)
}
// ── Lesson lifecycle: staleness sweep ──────────────────────────────
const STALE_AFTER_DAYS: i64 = 60;
pub fn run_staleness_sweep(memory_dir: &std::path::Path) -> std::io::Result<Vec<String>> {
@@ -572,7 +531,6 @@ pub fn run_staleness_sweep(memory_dir: &std::path::Path) -> std::io::Result<Vec<
pub fn maybe_run_staleness_sweep(state: &mut AppStateRest) {
let now = chrono::Utc::now().timestamp_millis();
// Only run every 10 minutes at most.
if now.saturating_sub(state.misc.last_staleness_sweep_ms) < 600_000 {
return;
}
@@ -587,26 +545,16 @@ pub fn maybe_run_staleness_sweep(state: &mut AppStateRest) {
}
}
// ── Contradiction detection ────────────────────────────────────────
//
// A cheap, no-embedding heuristic: split the new text into normalized
// directive phrases ("always use X", "never use Y", "prefer Z") and check
// for an existing lesson with the *opposite* directive on the same topic.
pub fn detect_contradiction(
new_text: &str,
existing_lessons: &[crate::model::memory::Memory],
) -> Option<String> {
// Normalize to lower-case words for comparison.
let new_words: std::collections::HashSet<String> = new_text
.to_lowercase()
.split(|c: char| !c.is_alphanumeric())
.filter(|w| w.len() >= 4 && !is_stop_word(w))
.map(|w| w.to_string())
.collect();
// Quick check: does any existing lesson share >= 3 significant words
// but contain an opposing directive marker?
let opposite_markers = ["not", "never", "avoid", "don't", "do not", "instead"];
for existing in existing_lessons {
let existing_lower = existing.content.to_lowercase();
@@ -618,7 +566,6 @@ pub fn detect_contradiction(
let shared = new_words.intersection(&exist_words).count();
if shared >= 3 {
// Same topic -- check for opposing directive.
let new_has_opposite = opposite_markers.iter().any(|m| new_text.to_lowercase().contains(m));
let old_has_opposite = opposite_markers.iter().any(|m| existing_lower.contains(m));
if new_has_opposite != old_has_opposite {
@@ -643,8 +590,6 @@ fn is_stop_word(w: &str) -> bool {
)
}
// ── Pending lesson calibration queue ───────────────────────────────
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PendingLesson {
pub lesson: Lesson,
@@ -675,13 +620,10 @@ pub fn add_pending_lesson(session_dir: &std::path::Path, lesson: Lesson, auto_re
});
save_pending_lessons(session_dir, &pending)
}
/// Process pending lessons: resolve auto-resolve ones (Auto/Yolo mode) after
/// a grace window, return ones that need explicit keypress.
pub fn process_pending_lessons(session_dir: &std::path::Path, memory_dir: &std::path::Path) -> std::io::Result<Vec<PendingLesson>> {
let pending = load_pending_lessons(session_dir);
let now = chrono::Utc::now().timestamp_millis();
let grace_window = 5_000; // 5 seconds in Auto/Yolo mode
let grace_window = 5_000;
let mut remaining = Vec::new();
let mut to_keep = Vec::new();
@@ -692,8 +634,6 @@ pub fn process_pending_lessons(session_dir: &std::path::Path, memory_dir: &std::
remaining.push(p.clone());
}
}
// Write kept lessons to memory
for lesson in &to_keep {
let mem = crate::model::memory::Memory {
name: lesson.name.clone(),
@@ -715,8 +655,6 @@ pub fn process_pending_lessons(session_dir: &std::path::Path, memory_dir: &std::
save_pending_lessons(session_dir, &remaining)?;
Ok(remaining)
}
/// Resolve a specific pending lesson (keep or discard).
pub fn resolve_pending_lesson(
session_dir: &std::path::Path,
memory_dir: &std::path::Path,
src/app/runtime/actions/mod.rs
-28
View File
@@ -58,16 +58,6 @@ pub enum Action {
LessonReject {
name: String,
},
#[expect(dead_code)]
RecordUsage {
tokens_in: u64,
tokens_out: u64,
duration_ms: u64,
},
#[expect(dead_code)]
RecordReviewTokens {
tokens: u64,
},
SaveSession,
ResumeSession,
RefreshSessions,
@@ -333,7 +323,6 @@ pub fn apply_action(state: &mut AppStateRest, action: Action) {
Action::Tick => {
let now_ms = chrono::Utc::now().timestamp_millis();
state.misc.drain_expired_toasts(now_ms);
// Idle-time housekeeping.
crate::app::review::maybe_run_staleness_sweep(state);
if let Some(ref rt) = state.session_runtime {
let _ = crate::app::review::process_pending_lessons(&rt.session_dir, &state.memory_dir);
@@ -385,10 +374,7 @@ pub fn apply_action(state: &mut AppStateRest, action: Action) {
let _ = trigger_review(state);
}
} else if kind == "review" {
// Track review outcome: check if lessons were found.
// Format: "Quality review: <verdict> [N lesson(s)]"
let lessons_found = if message.contains("lesson") || message.contains("Lesson") {
// Check for "N lesson(s)" pattern at end
message.rsplit(' ').next().and_then(|w| {
w.trim_end_matches(')').trim_end_matches('s')
.split('(').next_back()
@@ -426,18 +412,6 @@ pub fn apply_action(state: &mut AppStateRest, action: Action) {
state.dirty = true;
}
}
Action::RecordUsage { tokens_in, tokens_out, duration_ms } => {
if let Some(ref mut rt) = state.session_runtime {
rt.record_api_call(tokens_in, tokens_out, duration_ms);
}
state.dirty = true;
}
Action::RecordReviewTokens { tokens } => {
if let Some(ref mut rt) = state.session_runtime {
rt.record_review_tokens(tokens);
}
state.dirty = true;
}
Action::LessonAccept { name } => {
if let Some(ref rt) = state.session_runtime {
let _ = crate::app::review::resolve_pending_lesson(
@@ -729,11 +703,9 @@ fn auto_create_retrospective(state: &mut AppStateRest) {
}
Ok(None) => {}
Err(e) => {
// Silently handle — retrospective is best-effort.
let _ = e;
}
}
// Attempt consensus promotion for global-scope lessons.
let lessons: Vec<crate::model::memory::Memory> = crate::model::memory::Memory::list(&state.memory_dir)
.iter()
.filter_map(|n| crate::model::memory::Memory::read(&state.memory_dir, n).ok())
src/app/runtime/mod.rs
-2
View File
@@ -1,6 +1,4 @@
pub mod actions;
pub mod commands;
pub mod event_loop;
#[expect(dead_code)]
pub mod shortsend;
pub mod stream;
src/app/runtime/stream/tools/mod.rs
-3
View File
@@ -1,3 +0,0 @@
// Tool execution dispatch — superseded by inline per-tool call in
// app::runtime::actions::execute_one_tool within the SubmitInput loop.
// This module is preserved as a placeholder.
src/app/runtime/stream/turn.rs
-4
View File
@@ -1,4 +0,0 @@
// Stream module — superseded by the inline tool-calling loop in
// app::runtime::actions (Action::SubmitInput / Tick pipeline).
// This module is preserved as a placeholder; all previous content
// has been removed since it duplicated logic now in actions/mod.rs.
src/app/sec/daemon.rs
+207 -8
View File
@@ -1,29 +1,228 @@
use anyhow::Result;
use std::io::{BufRead, BufReader, Write};
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Instant;
use anyhow::{Context, Result, anyhow};
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SecResponse {
pub id: String,
pub ok: bool,
#[serde(default)]
pub output: String,
#[serde(default)]
pub error: String,
#[serde(default)]
pub duration_ms: u64,
#[serde(default)]
pub ts: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthResult {
pub tools: std::collections::HashMap<String, ToolHealth>,
pub available_count: usize,
pub total_count: usize,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ToolHealth {
pub available: bool,
}
pub struct SecDaemon {
pub pid: Option<u32>,
pub running: bool,
child: Option<Child>,
child_stdin: Option<Mutex<Box<dyn Write + Send>>>,
response_buf: Arc<Mutex<Vec<String>>>,
running: Arc<AtomicBool>,
token: String,
next_req_id: Arc<Mutex<u64>>,
}
impl SecDaemon {
pub fn new() -> Self {
SecDaemon {
pid: None,
running: false,
child: None,
child_stdin: None,
response_buf: Arc::new(Mutex::new(Vec::new())),
running: Arc::new(AtomicBool::new(false)),
token: uuid::Uuid::new_v4().to_string(),
next_req_id: Arc::new(Mutex::new(1)),
}
}
pub fn start(&mut self) -> Result<()> {
self.running = true;
if self.running.load(Ordering::SeqCst) {
return Ok(());
}
let mut child = Command::new("python3")
.arg("-m")
.arg("zesdex_sec_daemon")
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.context("failed to spawn security daemon (python3 on PATH?)")?;
let child_stdin = child.stdin.take()
.ok_or_else(|| anyhow!("no stdin"))?;
let child_stdout = child.stdout.take()
.ok_or_else(|| anyhow!("no stdout"))?;
let resp_buf = self.response_buf.clone();
let running = self.running.clone();
std::thread::spawn(move || {
let mut reader = BufReader::new(child_stdout);
loop {
if !running.load(Ordering::SeqCst) {
break;
}
let mut line = String::new();
match reader.read_line(&mut line) {
Ok(0) => break,
Ok(_) => {
if let Ok(mut buf) = resp_buf.lock() {
buf.push(line.trim().to_string());
}
}
Err(_) => break,
}
}
});
{
let mut stdin = Box::new(child_stdin) as Box<dyn Write + Send>;
let handshake = serde_json::json!({"op": "handshake", "token": self.token});
writeln!(stdin, "{}", handshake).context("handshake write failed")?;
stdin.flush()?;
self.child_stdin = Some(Mutex::new(stdin));
}
self.running.store(true, Ordering::SeqCst);
self.child = Some(child);
Ok(())
}
pub fn stop(&mut self) -> Result<()> {
self.running = false;
self.running.store(false, Ordering::SeqCst);
self.child_stdin = None;
if let Some(mut child) = self.child.take() {
child.kill().ok();
child.wait().ok();
}
Ok(())
}
pub fn is_running(&self) -> bool {
self.running.load(Ordering::SeqCst)
}
fn next_id(&self) -> String {
let mut id = self.next_req_id.lock().unwrap();
*id += 1;
format!("sec-{}", id)
}
fn do_call(&self, request: Value, timeout_ms: u64) -> Result<SecResponse> {
if !self.running.load(Ordering::SeqCst) {
return Err(anyhow!("security daemon is not running"));
}
let req_id = request.get("id")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow!("missing id in request"))?
.to_string();
let stdin_lock = self.child_stdin.as_ref()
.ok_or_else(|| anyhow!("stdin not available"))?;
let mut stdin = stdin_lock.lock().map_err(|_| anyhow!("stdin lock"))?;
writeln!(stdin, "{}", serde_json::to_string(&request)?)
.context("write request")?;
stdin.flush()?;
drop(stdin);
let start = Instant::now();
let buf = self.response_buf.clone();
loop {
if start.elapsed().as_millis() as u64 > timeout_ms {
return Err(anyhow!("call timed out after {}ms", timeout_ms));
}
{
let mut buf_lock = buf.lock().map_err(|_| anyhow!("buf lock"))?;
if let Some(pos) = buf_lock.iter().position(|l| {
serde_json::from_str::<SecResponse>(l)
.ok()
.map(|r| r.id == req_id)
.unwrap_or(false)
}) {
let line = buf_lock.remove(pos);
return serde_json::from_str(&line)
.map_err(|e| anyhow!("parse response: {}", e));
}
}
std::thread::sleep(std::time::Duration::from_millis(10));
}
}
pub fn call(&self, tool: &str, args: Value, timeout_ms: u64) -> Result<String> {
let request = serde_json::json!({
"id": self.next_id(),
"op": "call",
"tool": tool,
"args": args,
"timeout": timeout_ms,
});
let resp = self.do_call(request, timeout_ms)?;
if resp.ok {
Ok(resp.output)
} else {
Err(anyhow!("{}", resp.error))
}
}
pub fn health_check(&self) -> Result<HealthResult> {
let request = serde_json::json!({
"id": self.next_id(),
"op": "health",
});
let resp = self.do_call(request, 10_000)?;
if resp.ok {
serde_json::from_str(&resp.output)
.map_err(|e| anyhow!("parse health: {}", e))
} else {
Err(anyhow!("health check failed: {}", resp.error))
}
}
pub fn install_tool(&self, tool_name: &str) -> Result<String> {
let request = serde_json::json!({
"id": self.next_id(),
"op": "install",
"tool": tool_name,
"timeout": 120_000,
});
let resp = self.do_call(request, 120_000)?;
if resp.ok {
Ok(resp.output)
} else {
Err(anyhow!("install failed: {}", resp.error))
}
}
pub fn pid(&self) -> Option<u32> {
self.child.as_ref().map(|c| c.id())
}
}
impl Drop for SecDaemon {
fn drop(&mut self) {
let _ = self.stop();
}
}
pub fn health_check() -> Result<bool> {
Ok(true)
let path = crate::security::install::get_sidecar_path();
Ok(path.exists())
}
src/app/state/misc.rs
+4
View File
@@ -169,6 +169,8 @@ pub struct MiscState {
pub security_acknowledged: bool,
pub esc_press_count: u32,
pub last_staleness_sweep_ms: i64,
pub effort_level: usize,
pub editor: Option<crate::app::mode::editor::EditorState>,
}
impl MiscState {
@@ -182,6 +184,8 @@ impl MiscState {
security_acknowledged: false,
esc_press_count: 0,
last_staleness_sweep_ms: 0,
effort_level: 1,
editor: None,
}
}
src/app/workflow/engine.rs
-21
View File
@@ -110,29 +110,8 @@ pub fn run_workflow(script: &WorkflowScript, args: &HashMap<String, String>) ->
Ok("workflow completed".to_string())
}
#[expect(dead_code)]
pub fn push_finding(engine: &mut WorkflowEngine, text: &str) {
engine.findings.push(text.to_string());
}
pub fn note_finding(text: &str) {
if let Ok(mut findings) = FINDINGS.lock() {
findings.push(text.to_string());
}
}
#[expect(dead_code)]
pub fn current_findings() -> Vec<String> {
if let Ok(findings) = FINDINGS.lock() {
findings.clone()
} else {
Vec::new()
}
}
#[expect(dead_code)]
pub fn clear_findings() {
if let Ok(mut findings) = FINDINGS.lock() {
findings.clear();
}
}
src/main.rs
-6
View File
@@ -82,8 +82,6 @@ fn run_single_process() -> Result<()> {
.internet_mode(state.settings.internet_mode.clone())
.origin(crate::app::state::types::Origin::Main)
.build();
// Read ToolCtx unused fields
let _ = &ctx.session_dir;
let _ = &ctx.memory_dir;
let _ = &ctx.download_dir;
@@ -108,8 +106,6 @@ fn run_single_process() -> Result<()> {
let _ = tool::DEFERRED_TOOLS;
let _ = tool::fs::helpers::arg_str(&serde_json::json!({"test": "value"}), "test");
let _ = tool::fs::helpers::not_found_help(&ctx, std::path::Path::new("/nonexistent"), "test");
// shell_filter function references
let _ = tool::shell_filter::credentials::check_credential_read("echo safe");
let _ = tool::shell_filter::git::check_git_destructive("git push");
let _ = tool::shell_filter::git::check_git_destructive("git status");
@@ -533,7 +529,6 @@ fn run_attach(session_id: &str) -> Result<()> {
Ok(())
}
fn run_loop(
state: &mut app::state::rest::AppStateRest,
terminal: &mut Terminal<CrosstermBackend<io::Stdout>>,
@@ -548,7 +543,6 @@ fn run_loop(
result
}
fn run_loop_inner(
state: &mut app::state::rest::AppStateRest,
terminal: &mut Terminal<CrosstermBackend<io::Stdout>>,
src/model/memory.rs
-16
View File
@@ -158,25 +158,14 @@ pub fn export_lessons(memory_dir: &Path, output: &Path) -> std::io::Result<()> {
std::fs::write(output, data)?;
Ok(())
}
/// Promote a lesson to global scope, requiring consensus.
/// Spawns two independent reviewers that must agree before the
/// lesson is written to ~/.zesdex/memory/.
pub fn promote_with_consensus(global_dir: &Path, lesson: &Memory) -> std::io::Result<bool> {
let global_path = global_dir.join("memory");
std::fs::create_dir_all(&global_path)?;
// Check if already in global store.
let existing = Memory::list(&global_path);
let slug = Memory::slugify(&lesson.name).unwrap_or_default();
if existing.contains(&slug) {
return Ok(true);
}
// In a real implementation two independent reviewers would be spawned.
// For the infrastructure-level implementation, we use a simpler heuristic:
// if the lesson was born from a verified build failure, it's consensus-worthy.
// Otherwise, require an explicit human calibration.
let consensus = lesson.outcome.as_deref() == Some("verified");
if consensus {
@@ -204,22 +193,17 @@ pub fn import_lessons(memory_dir: &Path, input: &Path) -> std::io::Result<usize>
}
Ok(imported)
}
/// Automatically create a retrospective for a session that has been
/// active for at least 60 seconds and has edits or lessons.
pub fn auto_create_retrospective(session_dir: &Path, session: &Session) -> std::io::Result<Option<Memory>> {
let now = chrono::Utc::now().timestamp_millis();
let session_age_ms = now.saturating_sub(session.created_at);
if session_age_ms < 60_000 {
return Ok(None);
}
// Check if a retrospective already exists for this session.
let retro_name = format!("retrospective-{}", session.id);
let retro_path = Memory::path(session_dir, &retro_name);
if retro_path.exists() {
return Ok(None);
}
// Collect lessons per-session-dir memory store.
let lessons: Vec<Memory> = Memory::list(session_dir)
.iter()
.filter_map(|n| Memory::read(session_dir, n).ok())
src/service/oauth/mod.rs
-10
View File
@@ -1,13 +1,3 @@
#[expect(dead_code)]
pub mod pkce;
#[expect(dead_code)]
pub mod loopback;
#[expect(dead_code)]
pub mod manager;
#[expect(unused_imports)]
pub use manager::{OAuthManager, OAuthConfig};
#[expect(unused_imports)]
pub use pkce::CodeVerifier;
#[expect(unused_imports)]
pub use loopback::LoopbackServer;
src/tool/mod.rs
-1
View File
@@ -96,7 +96,6 @@ impl ToolCtxBuilder {
pub fn worktrees_dir(mut self, v: PathBuf) -> Self { self.worktrees_dir = v; self }
pub fn internet_mode(mut self, v: super::model::settings::InternetMode) -> Self { self.internet_mode = v; self }
pub fn origin(mut self, v: crate::app::state::types::Origin) -> Self { self.origin = v; self }
#[expect(dead_code)]
pub fn graduated_checks(mut self, v: Vec<GraduatedCheck>) -> Self { self.graduated_checks = v; self }
pub fn build(self) -> ToolCtx {
ToolCtx {