agentsdock_publish.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Publish files to the active AgentsDock chat turn and verify the receipt."""
from __future__ import annotations
import argparse
import ipaddress
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
import uuid
from pathlib import Path
from typing import Any
PROVIDER_RUNTIME_VALUE_MAX_BYTES = 4096
class PublishCLIError(RuntimeError):
"""A concise user-facing publication failure."""
class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
"""Keep the privileged publication request on the validated loopback URL."""
def redirect_request(self, *_args: Any, **_kwargs: Any) -> None:
return None
def nonempty_chat_id(value: str) -> str:
chat_id = value.strip()
if not chat_id:
raise argparse.ArgumentTypeError("--chat-id must not be empty")
return chat_id
def host_is_loopback(host: str) -> bool:
try:
address = ipaddress.ip_address(host)
except ValueError:
return host.lower() == "localhost"
if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped:
address = address.ipv4_mapped
return address.is_loopback
def canonical_http_origin(value: str, label: str) -> tuple[str, bool]:
raw = value.strip()
try:
parsed = urllib.parse.urlsplit(raw)
port = parsed.port or 80
except ValueError as exc:
raise PublishCLIError(f"{label} must be an HTTP origin") from exc
if (
parsed.scheme.lower() != "http"
or not parsed.hostname
or parsed.username is not None
or parsed.password is not None
or parsed.path not in {"", "/"}
or parsed.query
or parsed.fragment
):
raise PublishCLIError(f"{label} must be an HTTP origin")
host = parsed.hostname.lower()
try:
address = ipaddress.ip_address(host)
if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped:
address = address.ipv4_mapped
host = address.compressed
loopback = address.is_loopback
url_host = f"[{host}]" if isinstance(address, ipaddress.IPv6Address) else host
except ValueError:
loopback = host == "localhost"
url_host = host
return f"http://{url_host}:{port}", loopback
def validated_server_url(authority_origin: str = "") -> str:
raw_server_url = os.environ.get("AGENTSDOCK_SERVER_URL", "").strip()
if not raw_server_url:
raise PublishCLIError("missing agent environment: AGENTSDOCK_SERVER_URL")
server_origin, loopback = canonical_http_origin(
raw_server_url,
"AGENTSDOCK_SERVER_URL",
)
runtime_origin = bounded_identity_value(
os.environ.get("AGENTSDOCK_PROVIDER_SERVER_ORIGIN"),
"AGENTSDOCK_PROVIDER_SERVER_ORIGIN",
)
if runtime_origin:
canonical_runtime, _runtime_loopback = canonical_http_origin(
runtime_origin,
"AGENTSDOCK_PROVIDER_SERVER_ORIGIN",
)
if canonical_runtime != server_origin:
raise PublishCLIError(
"AGENTSDOCK_SERVER_URL conflicts with the live provider origin"
)
if loopback:
return raw_server_url.rstrip("/")
if not authority_origin:
raise PublishCLIError(
"non-loopback AGENTSDOCK_SERVER_URL must match the authority origin"
)
canonical_authority, _authority_loopback = canonical_http_origin(
authority_origin,
"authority provider_server_origin",
)
if canonical_authority != server_origin:
raise PublishCLIError(
"non-loopback AGENTSDOCK_SERVER_URL must match the authority origin"
)
return server_origin
def loopback_server_url() -> str:
"""Compatibility validator for legacy explicit loopback CLI calls."""
return validated_server_url()
def bounded_identity_value(value: str | None, label: str) -> str:
clean = str(value or "").strip()
try:
size = len(clean.encode("utf-8"))
except UnicodeEncodeError as exc:
raise PublishCLIError(f"{label} is not valid UTF-8") from exc
if size > PROVIDER_RUNTIME_VALUE_MAX_BYTES:
raise PublishCLIError(f"{label} exceeds the provider runtime limit")
return clean
def selected_authority_path(authority_file: str | None) -> Path:
explicit = bounded_identity_value(authority_file, "--authority-file")
ambient = bounded_identity_value(
os.environ.get("AGENTSDOCK_PROVIDER_AUTHORITY_FILE"),
"AGENTSDOCK_PROVIDER_AUTHORITY_FILE",
)
if explicit and ambient:
explicit_key = os.path.abspath(os.path.expanduser(explicit))
ambient_key = os.path.abspath(os.path.expanduser(ambient))
if explicit_key != ambient_key:
raise PublishCLIError(
"--authority-file conflicts with the live provider authority"
)
selected = explicit or ambient
if not selected:
raise PublishCLIError("--authority-file is required")
return Path(selected).expanduser()
def provider_authority(authority_file: str | None) -> tuple[str, str]:
path = selected_authority_path(authority_file)
try:
if path.stat().st_mode & 0o077:
raise PublishCLIError("authority file permissions are unsafe")
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise PublishCLIError(f"could not read authority file: {exc}") from exc
capability = str(payload.get("provider_capability") or payload.get("capability") or "")
source_session_id = str(payload.get("source_session_id") or "").strip()
if not capability or not source_session_id:
raise PublishCLIError("authority file is invalid")
return capability, source_session_id
def authority_server_origin(authority_file: str | None) -> str:
path = selected_authority_path(authority_file)
try:
if path.stat().st_mode & 0o077:
raise PublishCLIError("authority file permissions are unsafe")
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise PublishCLIError(f"could not read authority file: {exc}") from exc
return bounded_identity_value(
payload.get("provider_server_origin"),
"authority provider_server_origin",
)
def requested_chat_scope(
chat_id: str | None,
authority_chat_id: str,
) -> str:
explicit = bounded_identity_value(chat_id, "--chat-id")
ambient = bounded_identity_value(
os.environ.get("AGENTSDOCK_CHAT_ID"),
"AGENTSDOCK_CHAT_ID",
)
if explicit and ambient and explicit != ambient:
raise PublishCLIError("--chat-id conflicts with AGENTSDOCK_CHAT_ID")
for candidate in (explicit, ambient):
if candidate and candidate != authority_chat_id:
raise PublishCLIError("--chat-id does not match the authority file")
return authority_chat_id
def load_manifest(path: str) -> list[Any]:
try:
data = json.loads(Path(path).read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise PublishCLIError(f"could not read manifest {path!r}: {exc}") from exc
if not isinstance(data, dict) or not isinstance(data.get("files"), list):
raise PublishCLIError("manifest must be a JSON object with a files array")
return list(data["files"])
def parse_entry_json(value: str) -> dict[str, Any]:
try:
entry = json.loads(value)
except json.JSONDecodeError as exc:
raise argparse.ArgumentTypeError(f"invalid --entry-json: {exc}") from exc
if not isinstance(entry, dict):
raise argparse.ArgumentTypeError("--entry-json must decode to an object")
return entry
def requested_files(args: argparse.Namespace) -> list[Any]:
entries: list[Any] = list(args.paths)
entries.extend(args.entry_json)
if args.manifest:
entries.extend(load_manifest(args.manifest))
if not entries:
raise PublishCLIError("provide at least one absolute file path")
return entries
def http_error_detail(raw: str) -> str:
try:
decoded = json.loads(raw)
except json.JSONDecodeError:
return raw
if isinstance(decoded, dict):
detail = decoded.get("detail")
if isinstance(detail, str):
return detail
if detail is not None:
return json.dumps(detail, separators=(",", ":"))
return json.dumps(decoded, separators=(",", ":"))
def publish(
chat_id: str | None,
files: list[Any],
*,
publication_id: str | None = None,
authority_file: str | None = None,
) -> dict[str, Any]:
capability, authority_chat_id = provider_authority(authority_file)
server_url = validated_server_url(authority_server_origin(authority_file))
chat_id = requested_chat_scope(chat_id, authority_chat_id)
publication_id = publication_id or f"pub_{uuid.uuid4().hex}"
encoded_chat_id = urllib.parse.quote(chat_id, safe="")
body = json.dumps({
"publication_id": publication_id,
"files": files,
}).encode("utf-8")
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
"X-AgentsDock-Provider-Capability": capability,
}
decoded: Any = None
ambiguous_error: BaseException | None = None
# urllib honors HTTP_PROXY even for loopback addresses on some hosts.
# Explicitly disable proxies and redirects so this token cannot leave the
# already-validated local endpoint.
opener = urllib.request.build_opener(
urllib.request.ProxyHandler({}),
NoRedirectHandler(),
)
for attempt in range(2):
attempt_headers = dict(headers)
if attempt:
attempt_headers["X-AgentsDock-Publication-Retry"] = "1"
request = urllib.request.Request(
f"{server_url}/api/agent/sessions/{encoded_chat_id}/artifacts",
data=body,
headers=attempt_headers,
method="POST",
)
try:
# Large videos can take time to copy from a network workspace. The
# server performs that copy outside its event loop and returns only
# after the artifact events are durable. One retry uses the same
# publication ID, so a lost response cannot create duplicate cards.
with opener.open(request, timeout=600) as response:
raw_response = response.read().decode("utf-8")
try:
decoded = json.loads(raw_response)
except json.JSONDecodeError as exc:
# A truncated response is ambiguous: the event batch may
# already be durable. Retry once with the same publication ID.
ambiguous_error = exc
if attempt == 0:
continue
break
ambiguous_error = None
break
except urllib.error.HTTPError as exc:
raw = exc.read().decode("utf-8", errors="replace")
detail = http_error_detail(raw)
raise PublishCLIError(
f"server rejected publication ({exc.code}): {detail or exc.reason}"
) from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
ambiguous_error = exc
if attempt == 0:
continue
if ambiguous_error is not None:
reason = getattr(ambiguous_error, "reason", ambiguous_error)
raise PublishCLIError(
f"could not confirm publication {publication_id}: {reason}"
) from ambiguous_error
if not isinstance(decoded, dict) or decoded.get("ok") is not True:
raise PublishCLIError("AgentsServer did not confirm publication")
if decoded.get("publication_id") != publication_id:
raise PublishCLIError("AgentsServer returned a receipt for another publication")
if decoded.get("chat_id") != chat_id:
raise PublishCLIError("AgentsServer returned a receipt for another chat")
run_id = decoded.get("run_id")
if not isinstance(run_id, str) or not run_id.strip():
raise PublishCLIError("AgentsServer returned a receipt without a run ID")
receipts = decoded.get("receipts")
if not isinstance(receipts, list) or len(receipts) != len(files):
raise PublishCLIError("AgentsServer returned an incomplete publication receipt")
artifact_ids: set[str] = set()
event_ids: set[str] = set()
for receipt in receipts:
artifact_id = str(receipt.get("artifact_id") or "") if isinstance(receipt, dict) else ""
event_id = str(receipt.get("event_id") or "") if isinstance(receipt, dict) else ""
if (
not isinstance(receipt, dict)
or not artifact_id.strip()
or not event_id.strip()
or not isinstance(receipt.get("event_seq"), int)
or isinstance(receipt.get("event_seq"), bool)
or int(receipt["event_seq"]) <= 0
or receipt.get("run_id") != run_id
or artifact_id in artifact_ids
or event_id in event_ids
):
raise PublishCLIError("AgentsServer returned an invalid publication receipt")
artifact_ids.add(artifact_id)
event_ids.add(event_id)
return decoded
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Attach files to the currently active AgentsDock chat turn.",
allow_abbrev=False,
)
parser.add_argument(
"--authority-file",
help="mode-0600 per-run AgentsDock provider authority file",
)
parser.add_argument(
"--chat-id",
type=nonempty_chat_id,
help="explicit chat scope (defaults to AGENTSDOCK_CHAT_ID)",
)
parser.add_argument(
"--manifest",
help="read additional entries from a legacy {\"files\": [...]} manifest",
)
parser.add_argument(
"--publication-id",
help="idempotency key (normally generated automatically)",
)
parser.add_argument(
"--entry-json",
action="append",
default=[],
type=parse_entry_json,
metavar="JSON",
help='publish an object entry such as {"path":"/tmp/demo.mp4","title":"Demo"}',
)
parser.add_argument("paths", nargs="*", help="absolute file paths")
return parser
def main(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
try:
result = publish(
args.chat_id,
requested_files(args),
publication_id=args.publication_id,
authority_file=args.authority_file,
)
except PublishCLIError as exc:
print(f"error: {exc}", file=sys.stderr)
return 1
print(json.dumps(result, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())