Merge develop into OCI input review

This commit is contained in:
Jonatan
2026-10-26 17:44:52 +01:00
155 changed files with 41949 additions and 162 deletions
@@ -1,7 +1,7 @@
"""Anthropic (Claude) provider implementation.
Anthropic's Claude models are excellent for text generation and translation.
Models use "-latest" aliases that auto-update to newest versions.
Models are addressed by their pinned IDs (the "-latest" aliases were retired).
"""
from typing import Optional, List
from .base import AIProvider, AIProviderError
@@ -15,29 +15,35 @@ class AnthropicProvider(AIProvider):
API_URL = "https://api.anthropic.com/v1/messages"
API_VERSION = "2023-06-01"
# Anthropic model aliases that resolve to pinned snapshots (per docs
# at platform.claude.com/docs/en/docs/about-claude/models). Kept as a
# runtime fallback for the `/api/notifications/provider-models`
# intersection with `verified_ai_models.json`; the JSON is the
# authoritative UI-facing list. Refreshed 2026-07-13 to drop the
# deprecated claude-3-5-* aliases (retired upstream, no longer in the
# Models API response) and add the current-generation IDs. Ordered
# cheapest → most capable so the first entry is a safe default.
# Current Claude API model IDs (per docs at
# platform.claude.com/docs/en/docs/about-claude/models). Kept as a
# runtime fallback for `/api/notifications/provider-models` when
# `verified_ai_models.json` has no Anthropic entry; the JSON is the
# authoritative UI-facing list. Refreshed 2026-10-26 against the live
# Models API. Ordered cheapest → most capable so the first entry is a
# safe default.
KNOWN_MODELS = [
"claude-haiku-4-5",
"claude-haiku-5-5",
"claude-haiku-4-5-20251001",
"claude-sonnet-5-5",
"claude-sonnet-5",
"claude-sonnet-4-6",
"claude-opus-5-5",
"claude-opus-4-8",
"claude-opus-4-7",
"claude-opus-4-6",
"claude-opus-4-5-20251101",
"claude-fable-5",
"claude-fable-5-1",
]
# Newer models (Opus 5.x, Fable) may think before answering and the
# thinking counts against max_tokens, so a small cap can end the turn
# with no text at all. max_tokens is a ceiling, not a cost.
MIN_MAX_TOKENS = 1024
def list_models(self) -> List[str]:
"""Return known Anthropic model aliases.
Anthropic doesn't have a public models list API, but their "-latest"
aliases auto-update to the newest versions, making them reliable choices.
"""
"""Return known Anthropic model IDs."""
return self.KNOWN_MODELS
def generate(self, system_prompt: str, user_message: str,
@@ -68,7 +74,7 @@ class AnthropicProvider(AIProvider):
'messages': [
{'role': 'user', 'content': user_message},
],
'max_tokens': max_tokens,
'max_tokens': max(max_tokens, self.MIN_MAX_TOKENS),
}
headers = {
@@ -79,11 +85,20 @@ class AnthropicProvider(AIProvider):
result = self._make_request(self.API_URL, payload, headers)
try:
# Anthropic returns content as array of content blocks
content = result['content']
if isinstance(content, list) and len(content) > 0:
return content[0].get('text', '').strip()
return str(content).strip()
except (KeyError, IndexError) as e:
raise AIProviderError(f"Unexpected response format: {e}")
# A safety classifier can decline even benign notifications
# (HTTP 200, stop_reason "refusal", empty content). Raising lets the
# caller fall back to the original, untranslated notification.
if result.get('stop_reason') == 'refusal':
category = (result.get('stop_details') or {}).get('category')
raise AIProviderError(f"Request declined by Anthropic (category: {category})")
# Content is a list of blocks; thinking blocks may come before the
# text, so take the first text block rather than content[0].
content = result.get('content')
if not isinstance(content, list):
raise AIProviderError("Unexpected response format: missing content")
for block in content:
if isinstance(block, dict) and block.get('type') == 'text':
return block.get('text', '').strip()
raise AIProviderError(
f"No text in response (stop_reason: {result.get('stop_reason')})")
+1
View File
@@ -148,6 +148,7 @@ cp "$SCRIPT_DIR/health_thresholds.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "
cp "$SCRIPT_DIR/managed_installs.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ managed_installs.py not found"
cp "$SCRIPT_DIR/lxc_update_observation.py" "$APP_DIR/usr/bin/"
cp "$SCRIPT_DIR/lxc_update_finalization.py" "$APP_DIR/usr/bin/"
cp "$SCRIPT_DIR/lxc_update_jobs.py" "$APP_DIR/usr/bin/"
cp "$SCRIPT_DIR/lxc_check_job.py" "$APP_DIR/usr/bin/"
cp "$SCRIPT_DIR/native_app_discovery.py" "$APP_DIR/usr/bin/"
cp "$SCRIPT_DIR/local_app_icons.py" "$APP_DIR/usr/bin/"
@@ -514,13 +514,10 @@ def get_provider_models():
'message': f'Found {len(models)} local models'
})
# Handle Anthropic - no models list API, return verified models directly
# Handle Anthropic - return verified models directly
if provider == 'anthropic':
models = list(verified_models) if verified_models else [
'claude-3-5-haiku-latest',
'claude-3-5-sonnet-latest',
'claude-3-opus-latest',
]
from ai_providers.anthropic_provider import AnthropicProvider
models = list(verified_models) if verified_models else list(AnthropicProvider.KNOWN_MODELS)
return jsonify({
'success': True,
'models': sorted(models),
+66
View File
@@ -134,6 +134,72 @@ def _run_script_completion_hook(script_path, params, exit_code, duration_seconds
except Exception as exc:
print(f"[ProxMenux] script completion hook failed: {exc}", flush=True)
@terminal_bp.route('/api/lxc-update-jobs', methods=['GET'])
@require_admin_scope
def lxc_update_jobs_status():
"""Read-only recovery after login. This never creates or restarts a job."""
from lxc_update_jobs import list_jobs
return jsonify({'jobs': list_jobs()})
@sock.route('/ws/lxc-update/<run_id>')
def lxc_update_viewer(ws, run_id):
"""An authenticated viewer, deliberately not the owner of the update."""
if not _ws_auth_check():
ws.close(reason=1008, message="Authentication required")
return
from lxc_update_jobs import get, start
try:
init = json.loads(ws.receive(timeout=10))
if not isinstance(init, dict) or type(init.get('resume')) is not bool:
raise ValueError("Invalid update job request")
offset = init.get('offset', 0)
if type(offset) is not int or offset < 0:
raise ValueError("Invalid update output cursor")
if init['resume']:
job = get(run_id, init.get('vmid'))
else:
job = start(run_id, init.get('script_path'), init.get('params'),
_run_script_completion_hook)
revision = -1
while True:
view = job.read(offset)
if view['text'] or view['offset'] != offset:
ws.send(json.dumps({'type': 'job_output', 'text': view['text'],
'offset': view['offset'], 'truncated': view['truncated']}))
offset = view['offset']
if view['revision'] != revision:
revision = view['revision']
# A second viewer must also dismiss a question already
# answered elsewhere, or cleared when the job ends.
ws.send(json.dumps({'type': 'web_interaction',
'interaction': view['interaction']}))
if view['state'] == 'complete':
ws.send(json.dumps({'type': 'script_complete', 'exit_code': view['exit_code']}))
break
data = ws.receive(timeout=0.2)
if data is None:
continue
try:
message = json.loads(data)
except (TypeError, ValueError):
message = data
if isinstance(message, dict) and message.get('type') == 'ping':
ws.send('{"type":"pong"}')
else:
job.input(message if isinstance(message, dict) else data)
except (ValueError, KeyError, OSError, TypeError) as exc:
try:
ws.send(json.dumps({'type': 'job_unavailable', 'message': str(exc)}))
except Exception:
pass
except Exception:
# Viewer loss, including auth expiry/reload, has no job side effects.
pass
# No process termination, PTY close or completion callback here. The job's
# worker alone owns those, even while no viewer is authenticated/connected.
@terminal_bp.route('/api/terminal/health', methods=['GET'])
def terminal_health():
"""Health check for terminal service"""
+375
View File
@@ -0,0 +1,375 @@
"""Browser-independent owners for explicitly confirmed LXC Updates runs.
Only the two Updates entrypoints are allowed. Import/list/attach never runs
anything. Interactive /ws/script consumers retain their original lifecycle.
Jobs survive viewer loss, not a Monitor/host restart; missing resume IDs fail
closed. The update shell still owns the shared scheduler/control/check flock.
"""
import codecs
import errno
import json
import os
import pty
import re
import select
import signal
import struct
import subprocess
import tempfile
import termios
import fcntl
import threading
import time
SCRIPTS = {
"/usr/local/share/proxmenux/scripts/lxc/apply_updates.sh",
"/usr/local/share/proxmenux/scripts/lxc/install_checked_os_updates.sh",
}
PARAMS = {
"VMID", "RUN_ID", "CT_NAME", "TARGET", "REQUESTED_TARGETS_JSON",
"TARGET_LABELS_JSON", "BACKUP", "BACKUP_STORAGE", "RESTART",
"UPDATE_COMMAND", "APP_NAME", "RUN_HELPER", "ALLOW_HELPER_WITH_CUSTOM",
"DOCKER_STANDALONE_TARGETS", "UPDATE_DOCKER_ENGINE",
"REFRESH_DOCKER_INVENTORY",
}
RUN_ID = re.compile(r"^[A-Za-z0-9_-]{1,128}$")
INTERACTION_ID = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
MAX_RUNTIME = 4 * 3600
# The shared runner restores a guest it started in its EXIT trap. Its package
# observation and `pct shutdown --timeout 60` need time after SIGTERM, too.
STOP_GRACE = 300
RESULT_TTL = 6 * 3600
MAX_OUTPUT = 256 * 1024
MAX_INPUT = 64 * 1024
MAX_JOBS = 32
_lock = threading.RLock()
_jobs = {}
class JobConflict(ValueError):
pass
def _validate(run_id, script, params):
if not isinstance(run_id, str) or not RUN_ID.fullmatch(run_id):
raise ValueError("Invalid update job ID")
if script not in SCRIPTS or os.path.realpath(script) != script:
raise ValueError("Only the LXC Updates entrypoints support background runs")
if not isinstance(params, dict) or set(params) - PARAMS:
raise ValueError("Invalid update parameters")
if any(not isinstance(v, str) or len(v) > 65536 for v in params.values()):
raise ValueError("Invalid update parameter value")
if len(json.dumps(params)) > 128 * 1024:
raise ValueError("Update parameters are too large")
if params.get("RUN_ID") != run_id:
raise ValueError("Update run ID does not match")
vmid = params.get("VMID", "")
if not re.fullmatch(r"[1-9][0-9]{0,8}", vmid):
raise ValueError("Invalid container ID")
if params.get("TARGET") not in ("os", "app", "both"):
raise ValueError("Invalid update target")
return int(vmid)
class UpdateJob:
def __init__(self, run_id, script, params, callback):
self.id, self.script, self.params = run_id, script, dict(params)
self.vmid = int(params["VMID"])
self.created_at = time.time()
self.started = time.monotonic()
self.finished_at = None
self.exit_code = None
self.state = "running"
self.output = ""
self.offset = 0
self.interaction = None
self.revision = 0
self.lock = threading.RLock()
self.callback = callback
self.stop_reason = None
self.stop_at = None
self.master = None
self.process = None
self.log_path = None
self._interaction_buffer = ""
self._input = bytearray()
def launch(self):
master, slave = pty.openpty()
self.master = master
try:
# A viewer's paste must never block the worker's output, timeout
# or cancellation when the updater does not read its stdin.
os.set_blocking(master, False)
fd, self.log_path = tempfile.mkstemp(prefix="proxmenux_update_", suffix=".log")
os.close(fd)
fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack("HHHH", 30, 120, 0, 0))
env = dict(os.environ, **self.params, EXECUTION_MODE="web",
WEB_LOG=self.log_path, TERM="xterm-256color", PYTHONUNBUFFERED="1")
self.process = subprocess.Popen(
["/bin/bash", self.script], stdin=slave, stdout=slave, stderr=slave,
start_new_session=True, env=env,
)
except BaseException:
os.close(master)
self.master = None
if self.log_path:
os.unlink(self.log_path)
raise
finally:
os.close(slave)
try:
threading.Thread(target=self._run, daemon=True,
name=f"lxc-update-{self.id}").start()
except BaseException:
# A failed worker start must not leave an untracked updater alive.
try:
os.killpg(self.process.pid, signal.SIGKILL)
except ProcessLookupError:
pass
self.process.wait()
os.close(master)
self.master = None
os.unlink(self.log_path)
raise
def _append(self, text):
with self.lock:
self.output = (self.output + text)[-MAX_OUTPUT:]
self.offset += len(text)
def _interactions(self, stream):
self._interaction_buffer += stream.read(16384)
if len(self._interaction_buffer) > 65536:
self._interaction_buffer = ""
return
lines = self._interaction_buffer.split("\n")
self._interaction_buffer = lines.pop()
for line in lines:
if not line.startswith("WEB_INTERACTION:"):
continue
try:
import base64
parts = line.rstrip("\n")[16:].split(":", 4)
kind, identity, title, message = parts[:4]
if not INTERACTION_ID.fullmatch(identity):
continue
value = dict(type=kind, id=identity,
title=base64.b64decode(title).decode(),
message=base64.b64decode(message).decode())
if len(parts) > 4:
if kind == "menu":
value["options"] = json.loads(parts[4])
elif kind in ("input", "inputbox"):
value["default"] = base64.b64decode(parts[4]).decode()
with self.lock:
if not self.stop_reason:
self.interaction = value
self.revision += 1
except (ValueError, UnicodeError, IndexError):
continue
def _signal(self, reason):
with self.lock:
if self.process.poll() is not None or self.stop_reason:
return
self.stop_reason, self.stop_at = reason, time.monotonic()
self._input.clear()
self.interaction = None
self.revision += 1
try:
os.killpg(self.process.pid, signal.SIGTERM)
except ProcessLookupError:
pass
def cancel(self):
self._signal("cancelled")
def _write_input(self):
# The descriptor is nonblocking, so even a full PTY cannot hold this
# lock. Retain unwritten bytes in order until the child can read them.
with self.lock:
if not self._input or self.stop_reason:
return
try:
written = os.write(self.master, self._input[:16384])
except BlockingIOError:
return
except OSError as exc:
if exc.errno != errno.EIO:
raise
# The child can close its terminal and keep working. Losing
# stdin must not turn that update into a viewer failure.
self._input.clear()
return
del self._input[:written]
def _run(self):
decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
eof, exited_at = False, None
try:
with open(self.log_path, encoding="utf-8", errors="replace") as interactions:
while True:
now = time.monotonic()
running = self.process.poll() is None
if not running:
exited_at = exited_at or now
if eof or now - exited_at >= 1:
break
if time.monotonic() - self.started >= MAX_RUNTIME:
self._signal("timeout")
if running and self.stop_at and now - self.stop_at >= STOP_GRACE:
try:
os.killpg(self.process.pid, signal.SIGKILL)
except ProcessLookupError:
pass
self._interactions(interactions)
with self.lock:
pending_input = bool(self._input) and running and not self.stop_reason
readable, writable, _ = select.select(
[] if eof else [self.master],
[self.master] if pending_input else [], [], 0.05)
if writable:
self._write_input()
if readable:
try:
data = os.read(self.master, 16384)
except BlockingIOError:
continue
except OSError as exc:
if exc.errno != errno.EIO:
raise
data = b""
if data:
self._append(decoder.decode(data))
continue
eof = True
code = self.process.wait()
self._append(decoder.decode(b"", final=True))
if self.stop_reason:
code = 124 if self.stop_reason == "timeout" else 130
except BaseException as exc:
self._append(f"\r\nUpdate viewer failed: {type(exc).__name__}\r\n")
self._signal("viewer failure")
try:
self.process.wait(timeout=STOP_GRACE)
except subprocess.TimeoutExpired:
os.killpg(self.process.pid, signal.SIGKILL)
self.process.wait()
code = 1
finally:
os.close(self.master)
self.master = None
try:
os.unlink(self.log_path)
except FileNotFoundError:
pass
with self.lock:
self.exit_code = code
self.state = "finalizing"
self.interaction = None
self._input.clear()
self.revision += 1
try:
self.callback(self.script, self.params, code, time.monotonic() - self.started)
finally:
with self.lock:
self.state = "complete"
self.finished_at = time.time()
self.revision += 1
def view(self):
with self.lock:
return dict(id=self.id, vmid=self.vmid, state=self.state,
target=self.params["TARGET"], ct_name=self.params.get("CT_NAME", ""),
created_at=self.created_at, finished_at=self.finished_at,
exit_code=self.exit_code, script_path=self.script)
def read(self, offset):
with self.lock:
begin = self.offset - len(self.output)
cursor = max(begin, min(offset, self.offset))
return dict(text=self.output[cursor - begin:], offset=self.offset,
truncated=offset < begin, interaction=self.interaction,
revision=self.revision, state=self.state, exit_code=self.exit_code)
def input(self, message):
with self.lock:
if self.state != "running" or self.master is None:
return
if isinstance(message, str):
if len(message) > 16384:
raise ValueError("Terminal input is too large")
if message == "\x03":
self.cancel()
else:
data = message.encode()
if len(self._input) + len(data) > MAX_INPUT:
raise ValueError("Terminal input buffer is full")
if not self.stop_reason:
self._input.extend(data)
elif message.get("type") == "cancel":
self.cancel()
elif message.get("type") == "resize":
cols = max(20, min(400, int(message.get("cols", 120))))
rows = max(5, min(200, int(message.get("rows", 30))))
fcntl.ioctl(self.master, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
elif message.get("type") == "interaction_response":
identity, value = message.get("id"), message.get("value")
if (not self.interaction or identity != self.interaction["id"]
or not isinstance(value, str) or len(value) > 16384):
raise ValueError("No matching pending interaction")
path = f"/tmp/proxmenux_response_{identity}"
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
with os.fdopen(fd, "w") as response:
response.write(value)
self.interaction = None
self.revision += 1
def start(run_id, script, params, callback):
vmid = _validate(run_id, script, params)
with _lock:
existing = _jobs.get(run_id)
if existing:
if existing.script != script or existing.params != params:
raise JobConflict("Update job ID already belongs to another request")
return existing
_prune()
if any(j.vmid == vmid and j.state != "complete" for j in _jobs.values()):
raise JobConflict("An update is already running for this container")
if len(_jobs) >= MAX_JOBS:
completed = [j for j in _jobs.values() if j.state == "complete"]
if not completed:
raise JobConflict("Too many active update jobs; try again later")
oldest = min(completed, key=lambda j: j.finished_at)
del _jobs[oldest.id]
job = UpdateJob(run_id, script, params, callback)
_jobs[run_id] = job
try:
job.launch()
except BaseException:
_jobs.pop(run_id, None)
raise
return job
def _prune():
now = time.time()
for identity, job in list(_jobs.items()):
if job.finished_at and now - job.finished_at > RESULT_TTL:
del _jobs[identity]
def get(run_id, vmid):
with _lock:
job = _jobs.get(run_id)
if job is None or job.vmid != int(vmid):
raise KeyError("Update job is unavailable; it was not restarted")
return job
def list_jobs():
with _lock:
_prune()
return [j.view() for j in sorted(_jobs.values(), key=lambda j: j.created_at, reverse=True)]
@@ -0,0 +1,65 @@
import sys
import unittest
from pathlib import Path
SCRIPTS_DIR = Path(__file__).resolve().parents[1]
if str(SCRIPTS_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPTS_DIR))
from ai_providers.anthropic_provider import AnthropicProvider
from ai_providers.base import AIProviderError
class CannedAnthropicProvider(AnthropicProvider):
def __init__(self, response):
super().__init__(api_key="test-key", model="claude-opus-5-5")
self.response = response
self.captured_payload = None
def _make_request(self, url, payload, headers):
self.captured_payload = payload
return self.response
class AnthropicProviderTests(unittest.TestCase):
def test_skips_thinking_block_before_text(self):
provider = CannedAnthropicProvider({
"stop_reason": "end_turn",
"content": [
{"type": "thinking", "thinking": "", "signature": "x"},
{"type": "text", "text": " Hola "},
],
})
self.assertEqual(provider.generate("system", "user"), "Hola")
def test_thinking_only_response_raises(self):
provider = CannedAnthropicProvider({
"stop_reason": "max_tokens",
"content": [{"type": "thinking", "thinking": "", "signature": "x"}],
})
with self.assertRaises(AIProviderError):
provider.generate("system", "user")
def test_refusal_with_empty_content_raises(self):
provider = CannedAnthropicProvider({
"stop_reason": "refusal",
"stop_details": {"type": "refusal", "category": "cyber"},
"content": [],
})
with self.assertRaisesRegex(AIProviderError, "cyber"):
provider.generate("system", "user")
def test_small_max_tokens_is_raised_to_floor(self):
provider = CannedAnthropicProvider({
"stop_reason": "end_turn",
"content": [{"type": "text", "text": "CONNECTION_OK"}],
})
provider.generate("system", "user", max_tokens=50)
self.assertEqual(provider.captured_payload["max_tokens"], AnthropicProvider.MIN_MAX_TOKENS)
provider.generate("system", "user", max_tokens=4000)
self.assertEqual(provider.captured_payload["max_tokens"], 4000)
if __name__ == "__main__":
unittest.main()