marek/creator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
Team3
2026-07-04 23:47:04 +02:00
parent 2d9ca00b47
commit 4105146c59
40 changed files with 2053 additions and 3208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
backend/agents.py
View File

@@ -22,13 +22,16 @@ import urllib.request
from contextlib import asynccontextmanager
from pathlib import Path
import httpx
from config import (PROVIDERS, DEFAULT_PROVIDER, MAX_CONCURRENT_AGENTS,
MAX_CONCURRENT_AGENTS_PER_TOPIC, MAX_CONCURRENT_INTERACTIVE,
resolve_role)
MAX_CONCURRENT_AGENTS_PER_TOPIC, MAX_CONCURRENT_API_AGENTS,
MAX_CONCURRENT_INTERACTIVE, RAM_MIN_FREE_PCT, resolve_role)
log = logging.getLogger("creator.agents")
_active_processes: dict[str, asyncio.subprocess.Process] = {}
_active_api: set[str] = set() # agent_keys of running direct-API calls (no process to kill)
_active_started: dict[str, float] = {} # agent_key → wall-clock start (for the live runtime display)
_active_labels: dict[str, str] = {} # agent_key → human-readable label (for display + events)
@@ -39,7 +42,8 @@ def active_agents(scope_prefix: str | None = None) -> list[dict]:
now = time.time()
out = [{"key": k, "label": _active_labels.get(k, ""), "runtime": round(now - t, 1)}
for k, t in list(_active_started.items())
if k in _active_processes and (not scope_prefix or k.startswith(scope_prefix))]
if (k in _active_processes or k in _active_api)
and (not scope_prefix or k.startswith(scope_prefix))]
return sorted(out, key=lambda a: -a["runtime"])
# Cancelled scopes (key prefixes, symmetric to kill_process). An agent whose
@@ -95,7 +99,8 @@ class _PrioritySemaphore:
self._value += 1
_batch_sem = _PrioritySemaphore(MAX_CONCURRENT_AGENTS)
_batch_sem = _PrioritySemaphore(MAX_CONCURRENT_AGENTS) # process tier (~310 MB RSS each)
_batch_sem_api = _PrioritySemaphore(MAX_CONCURRENT_API_AGENTS) # direct-API tier (~0 RAM)
_interactive_sem = asyncio.Semaphore(MAX_CONCURRENT_INTERACTIVE)
# Per-topic caps (lazily created): each topic gets its own priority semaphore of size
@@ -119,17 +124,19 @@ def _agent_priority(key: str) -> int:
@asynccontextmanager
async def _batch_gate(scope: str | None, priority: int):
async def _batch_gate(scope: str | None, priority: int, api: bool = False):
"""Per-topic slot FIRST (fair), then the GLOBAL slot by priority (earlier columns win when
agents are scarce). Order matters — a waiter holds only its per-topic slot while queueing globally."""
agents are scarce). Order matters — a waiter holds only its per-topic slot while queueing globally.
The per-topic cap is shared across both tiers; only the global cap is tiered (process vs API)."""
topic_sem = _topic_sems.setdefault(scope, _PrioritySemaphore(MAX_CONCURRENT_AGENTS_PER_TOPIC)) if scope else None
global_sem = _batch_sem_api if api else _batch_sem
if topic_sem is not None:
await topic_sem.acquire(priority)
await _batch_sem.acquire(priority)
await global_sem.acquire(priority)
try:
yield
finally:
_batch_sem.release()
global_sem.release()
if topic_sem is not None:
topic_sem.release()
@@ -151,6 +158,55 @@ async def _opencode_slot() -> None:
_opencode_next_start = start_at + _OPENCODE_START_DELAY
await asyncio.sleep(max(0.0, start_at - now))
# RAM-adaptive admission for opencode spawns (~310 MB RSS each): below RAM_MIN_FREE_PCT
# free memory new processes wait instead of starting. Gates admission only — running
# processes are never touched.
_RAM_GATE_FLOOR = 2 # below this many running: always admit (deadlock guard)
_RAM_PER_PROC_KB = 350 * 1024 # commit estimate: RSS ramps up slowly after spawn
_RAM_COMMIT_WINDOW_S = 10.0 # fresh admissions count as already-spent RAM
_RAM_POLL_S = 2.0
_opencode_running = 0 # opencode spawns only (stagger path), not claude
_opencode_recent_starts: list[float] = [] # monotonic timestamps of admissions
def _meminfo() -> tuple[int, int] | None:
"""(MemAvailable_kB, MemTotal_kB) from /proc/meminfo; None → gate fails open (non-Linux)."""
try:
text = Path("/proc/meminfo").read_text()
except OSError:
return None
m = {k: v for k, v in re.findall(r"^(MemTotal|MemAvailable):\s+(\d+)", text, re.MULTILINE)}
if "MemTotal" not in m or "MemAvailable" not in m:
return None
return int(m["MemAvailable"]), int(m["MemTotal"])
async def _ram_gate(agent_key: str) -> bool:
"""True = start admitted (commit registered), False = scope cancelled while waiting.
Check and commit-append happen in the same synchronous block (no await between) —
concurrent waiters on the loop cannot double-admit on the same free RAM."""
if RAM_MIN_FREE_PCT <= 0:
return True
waited = False
while True:
if _scope_cancelled(agent_key):
return False
mem = _meminfo()
if mem is None or _opencode_running < _RAM_GATE_FLOOR:
break # fail open / floor
avail_kb, total_kb = mem
now = time.monotonic()
_opencode_recent_starts[:] = [t for t in _opencode_recent_starts
if now - t < _RAM_COMMIT_WINDOW_S]
if avail_kb - len(_opencode_recent_starts) * _RAM_PER_PROC_KB >= total_kb * RAM_MIN_FREE_PCT / 100:
break
if not waited:
log.info("agent %s: RAM gate waiting (%.0f%% free)", agent_key, avail_kb * 100 / total_kb)
waited = True
await asyncio.sleep(_RAM_POLL_S)
_opencode_recent_starts.append(time.monotonic())
return True
_SLIM_CONFIG = Path(__file__).resolve().parent.parent / "dev-ops" / "opencode-slim.json"
# Capability → Claude --allowedTools
@@ -170,6 +226,17 @@ _OPENCODE_AGENTS = {
}
def _use_text_api(provider: str, model: str, capabilities: str, on_line) -> bool:
"""Direct API path only for tool-less, non-streaming MiniMax calls. Model-prefix check
instead of cli check: "lokal" (ollama) also runs via opencode. Missing key or
CREATOR_TEXT_API=0 (kill switch) falls back to the opencode process path."""
return (capabilities == "none" and on_line is None
and PROVIDERS[provider]["cli"] == "opencode"
and model.split("/", 1)[0] in ("minimax", "minimax-kalt")
and bool(os.environ.get("MINIMAX_API_KEY"))
and os.getenv("CREATOR_TEXT_API", "1") != "0")
def provider_available(provider: str) -> bool:
cfg = PROVIDERS.get(provider)
if not cfg:
@@ -258,10 +325,11 @@ async def run_agent(
provider, model = run_provider, PROVIDERS[run_provider].get(role, "")
if not model:
return 1, "", f"No model for role '{role}' (provider: {provider})"
if shutil.which(PROVIDERS[provider]["cli"]) is None:
use_api = _use_text_api(provider, model, capabilities, on_line)
if not use_api and shutil.which(PROVIDERS[provider]["cli"]) is None:
return 1, "", f"CLI '{PROVIDERS[provider]['cli']}' not installed (provider: {provider})"
queued = time.monotonic()
gate = _interactive_sem if lane == "interactive" else _batch_gate(scope, _agent_priority(agent_key))
gate = _interactive_sem if lane == "interactive" else _batch_gate(scope, _agent_priority(agent_key), api=use_api)
async with gate:
if _scope_cancelled(agent_key): # after the acquire: cancelled in the queue → no spawn
return 1, "", "cancelled"
@@ -270,9 +338,13 @@ async def run_agent(
status = "error"
rc = None
err_tail = ""
api_tokens = None
try:
log.info("agent %s: %s %s (role %s)", agent_key, provider, model, role)
if PROVIDERS[provider]["cli"] == "opencode":
if use_api:
rc_, out_, err_, api_tokens = await _run_text_api(agent_key, prompt, timeout, model, label=label)
res = (rc_, out_, err_)
elif PROVIDERS[provider]["cli"] == "opencode":
res = await _run_opencode(agent_key, prompt, timeout, provider, model, capabilities, on_line=on_line, label=label)
else:
res = await _run_claude_cli(agent_key, prompt, timeout, model, capabilities, label=label)
@@ -293,7 +365,9 @@ async def run_agent(
meta = {"provider": provider, "model": model, "role": role, "rc": rc}
if err_tail:
meta["stderr"] = err_tail
if PROVIDERS[provider]["cli"] == "opencode": # token accounting per agent
if api_tokens: # direct-API path: usage from the response, even on rc!=0
meta["tokens"] = api_tokens
elif PROVIDERS[provider]["cli"] == "opencode": # token accounting per agent
if (tok := await asyncio.to_thread(_session_tokens, agent_key)):
meta["tokens"] = tok
await on_event(topic=scope, kind="agent", key=agent_key, label=label,
@@ -316,9 +390,14 @@ async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None,
env=env,
)
global _opencode_running
if stagger:
await _opencode_slot() # spaced start slot; spawn itself is not serialized
if not await _ram_gate(agent_key):
return 1, "", "cancelled" # like the cancelled path in run_agent
await _opencode_slot() # gate BEFORE the start slot: an admission wave still gets spaced
process = await spawn()
if stagger:
_opencode_running += 1
# Collision-safe tracking: identical keys (e.g. same chunk label from parallel cards)
# get a ~n suffix — prefix-based kill/cancel still matches, nothing becomes an orphan.
track_key = agent_key
@@ -366,6 +445,8 @@ async def _communicate(agent_key: str, cmd: list[str], stdin_data: bytes | None,
)
return process.returncode, stdout.decode("utf-8", errors="replace"), stderr.decode("utf-8", errors="replace")
finally:
if stagger:
_opencode_running -= 1
# Pop only on identity: a slot restart under the same key must not evict
# the NEW process from tracking.
if _active_processes.get(track_key) is process:
@@ -441,6 +522,67 @@ async def _run_opencode(agent_key: str, prompt: str, timeout: int, provider: str
prompt_path.unlink(missing_ok=True)
# Direct text-API path (MiniMax, Anthropic Messages format): saves the ~310 MB RSS opencode
# process for tool-less calls. The native "minimax" and "minimax-kalt" opencode providers both
# resolve to this base URL — only the per-model options differ. Options source of truth:
# dev-ops/opencode.json (+ -slim); keep in sync on changes there.
_API_URL = "https://api.minimax.io/anthropic/v1/messages"
_API_VERSION = "2023-06-01"
_API_MAX_TOKENS = 32_000 # required Messages field; generate calls are long
_API_MODEL_OPTS = { # prefix "minimax" (native) → endpoint defaults (no entry)
"minimax-kalt/MiniMax-M3": {"temperature": 0.2, "thinking": {"type": "disabled"}},
"minimax-kalt/MiniMax-M2.7-highspeed": {"temperature": 0.3},
}
async def _run_text_api(agent_key: str, prompt: str, timeout: int, model: str,
label: str = "") -> tuple[int, str, str, dict | None]:
"""→ (rc, text, err, tokens). Tokens come straight from the response usage (same key set
as _session_tokens) and are returned even on rc!=0 — waste stays visible."""
body = {
"model": model.split("/", 1)[1], # without the opencode provider prefix
"max_tokens": _API_MAX_TOKENS,
"messages": [{"role": "user", "content": prompt}],
**_API_MODEL_OPTS.get(model, {}),
}
headers = {"x-api-key": os.environ.get("MINIMAX_API_KEY", ""), "anthropic-version": _API_VERSION}
track_key = agent_key
n = 2
while track_key in _active_api:
track_key = f"{agent_key}~{n}"
n += 1
_active_api.add(track_key)
_active_started[track_key] = time.time()
_active_labels[track_key] = label
start = time.monotonic()
try:
async with httpx.AsyncClient(timeout=httpx.Timeout(timeout, connect=30)) as client:
resp = await asyncio.wait_for( # belt: hard wall-clock cap like the process path
client.post(_API_URL, json=body, headers=headers), timeout=timeout)
except httpx.TimeoutException:
raise asyncio.TimeoutError # contract: run_agent/_race handle TimeoutError
except httpx.HTTPError as e:
return 1, "", f"{type(e).__name__}: {e}", None
finally:
_active_api.discard(track_key)
_active_started.pop(track_key, None)
_active_labels.pop(track_key, None)
log.info("agent %s: api done after %.1fs", agent_key, time.monotonic() - start)
if resp.status_code != 200:
return 1, "", f"HTTP {resp.status_code}: {resp.text[:300]}", None
data = resp.json()
# Only text blocks count — thinking blocks (native route) are skipped.
text = "".join(b.get("text", "") for b in data.get("content", []) if b.get("type") == "text")
u = data.get("usage") or {}
tokens = {"input": int(u.get("input_tokens") or 0), "output": int(u.get("output_tokens") or 0),
"reasoning": 0, "cache_read": int(u.get("cache_read_input_tokens") or 0),
"cache_write": int(u.get("cache_creation_input_tokens") or 0)}
if not text.strip():
return 1, "", f"empty response (stop_reason={data.get('stop_reason')})", tokens
err = "stop_reason=max_tokens (truncated)" if data.get("stop_reason") == "max_tokens" else ""
return 0, text, err, tokens
_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")