Chris Olson
07eb9a5411
- On a Claude Code session/usage limit ('resets 7:10pm'), _call_claude_cli now
pauses IN PLACE until the stated reset time + CLAUDE_CLI_LIMIT_BUFFER_SECONDS
(default 60s past), then retries — up to CLAUDE_CLI_LIMIT_MAX_WAITS windows.
Parses 12-hour reset times to their next occurrence; falls back to a fixed
wait when no time is present. Makes long unattended runs survive reset windows.
- load_library() reports invalid capability-set JSON as a clean SystemExit
(line number + reason) instead of a traceback; fixed a trailing-comma typo in
data/capability_sets/dx02.json.
- Docs: USAGE.md + .env.example document the claude-cli provider and its knobs.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
"""Capability Library loader + lightweight keyword pre-filter.
|
|
|
|
The design uses a vector ANN pre-filter to cut LLM calls ~60%. For the MVP we
|
|
approximate that with keyword overlap scoring (no embedding service required).
|
|
The interface is the same: given a signal, return candidate capabilities.
|
|
|
|
CAPABILITY SETS: the loop can target a subset of the portfolio at runtime.
|
|
- "ALL" (default): data/capabilities.json — the full portfolio.
|
|
- "<name>": data/capability_sets/<name>.json — a one-off targeted set
|
|
(same schema). Add new products by dropping another file in that folder and
|
|
passing `loop --capability <name>`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from functools import lru_cache
|
|
|
|
from . import config
|
|
|
|
# Selected via set_active_set() before the loop runs; "ALL" = full portfolio.
|
|
_active_set = "ALL"
|
|
|
|
|
|
def available_sets() -> list[str]:
|
|
sets_dir = config.DATA_DIR / "capability_sets"
|
|
names = sorted(p.stem for p in sets_dir.glob("*.json")) if sets_dir.exists() else []
|
|
return ["ALL"] + names
|
|
|
|
|
|
def set_active_set(name: str) -> None:
|
|
"""Select which capability library the loop reasons against."""
|
|
global _active_set
|
|
name = (name or "ALL").strip()
|
|
if name.upper() == "ALL":
|
|
_active_set = "ALL"
|
|
else:
|
|
path = config.DATA_DIR / "capability_sets" / f"{name}.json"
|
|
if not path.exists():
|
|
raise SystemExit(
|
|
f"Unknown capability set '{name}'. Available: {', '.join(available_sets())}"
|
|
)
|
|
_active_set = name
|
|
load_library.cache_clear()
|
|
|
|
|
|
def active_set() -> str:
|
|
return _active_set
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def load_library() -> dict:
|
|
path = (config.DATA_DIR / "capabilities.json" if _active_set == "ALL"
|
|
else config.DATA_DIR / "capability_sets" / f"{_active_set}.json")
|
|
try:
|
|
lib = json.loads(path.read_text())
|
|
except json.JSONDecodeError as e:
|
|
raise SystemExit(f"Capability set '{_active_set}' has invalid JSON "
|
|
f"({path.name}, line {e.lineno}): {e.msg}. Fix the file and re-run.")
|
|
if not lib.get("capabilities"):
|
|
raise SystemExit(f"Capability set '{_active_set}' ({path.name}) has no 'capabilities'.")
|
|
return lib
|
|
|
|
|
|
def capability_by_id(cap_id: str) -> dict | None:
|
|
for cap in load_library()["capabilities"]:
|
|
if cap["id"] == cap_id:
|
|
return cap
|
|
return None
|
|
|
|
|
|
def _tokens(text: str) -> set[str]:
|
|
return set(re.findall(r"[a-z]{3,}", text.lower()))
|
|
|
|
|
|
def prefilter(signal_content: str, top_k: int = 3) -> list[dict]:
|
|
"""Return capabilities whose keywords overlap the signal, best first.
|
|
|
|
Stand-in for the vector ANN pre-filter in the design (section 5.2).
|
|
In a targeted set (<=5 capabilities) the pre-filter is skipped: the whole
|
|
point of a targeted run is to evaluate EVERY signal against the chosen
|
|
product, and the LLM prompt stays small anyway.
|
|
"""
|
|
caps = load_library()["capabilities"]
|
|
if _active_set != "ALL" and len(caps) <= 5:
|
|
return caps
|
|
sig_tokens = _tokens(signal_content)
|
|
scored: list[tuple[float, dict]] = []
|
|
for cap in load_library()["capabilities"]:
|
|
kw = set(k.lower() for k in cap.get("keywords", []))
|
|
kw_tokens: set[str] = set()
|
|
for k in kw:
|
|
kw_tokens |= _tokens(k)
|
|
overlap = len(sig_tokens & kw_tokens)
|
|
if overlap:
|
|
scored.append((overlap, cap))
|
|
scored.sort(key=lambda x: x[0], reverse=True)
|
|
return [cap for _, cap in scored[:top_k]]
|
|
|
|
|
|
def library_summary_for_prompt(candidates: list[dict]) -> str:
|
|
"""Compact capability descriptions for the relevance-filter prompt."""
|
|
lines = []
|
|
for c in candidates:
|
|
lines.append(
|
|
f"- {c['id']} | {c['name']}: {c['description']}\n"
|
|
f" Buying signals: {'; '.join(c['buyingSignals'])}\n"
|
|
f" Anti-signals: {'; '.join(c['antiSignals'])}\n"
|
|
f" Target buyer: {', '.join(c['targetBuyer']['titles'])}"
|
|
)
|
|
return "\n".join(lines)
|