96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
"""Explicitly promote a small set of workspace files to user-facing artifacts."""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from core.artifacts import (
|
|
MAX_ARTIFACTS_PER_MESSAGE,
|
|
ArtifactPathError,
|
|
ArtifactRef,
|
|
ToolExecutionResult,
|
|
resolve_artifact_path,
|
|
)
|
|
|
|
from .base import Tool
|
|
|
|
|
|
class PublishArtifactsTool(Tool):
|
|
name = "publish_artifacts"
|
|
description = (
|
|
"Publish a small set of final deliverable files to the chat. Ordinary source, "
|
|
"temporary, intermediate, and project support files should stay in the file panel "
|
|
"and must not be published. Paths are relative to the current task working directory."
|
|
)
|
|
parameters = {
|
|
"type": "object",
|
|
"properties": {
|
|
"artifacts": {
|
|
"type": "array",
|
|
"minItems": 1,
|
|
"maxItems": MAX_ARTIFACTS_PER_MESSAGE,
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {
|
|
"type": "string",
|
|
"minLength": 1,
|
|
"maxLength": 1000,
|
|
"description": "File path relative to the current task working directory.",
|
|
},
|
|
"label": {
|
|
"type": "string",
|
|
"maxLength": 120,
|
|
"description": "Optional short user-facing label.",
|
|
},
|
|
},
|
|
"required": ["path"],
|
|
},
|
|
}
|
|
},
|
|
"required": ["artifacts"],
|
|
}
|
|
|
|
def __init__(self, working_dir: Path, **kwargs) -> None:
|
|
super().__init__(**kwargs)
|
|
self.working_dir = Path(working_dir)
|
|
|
|
def execute(self, artifacts: list[dict]) -> ToolExecutionResult | str:
|
|
if not isinstance(artifacts, list) or not artifacts:
|
|
return "[Error] artifacts must be a non-empty list"
|
|
if len(artifacts) > MAX_ARTIFACTS_PER_MESSAGE:
|
|
return f"[Error] at most {MAX_ARTIFACTS_PER_MESSAGE} artifacts may be published at once"
|
|
if self.user_root is None:
|
|
return "[Error] publish_artifacts requires a user workspace"
|
|
|
|
refs: list[ArtifactRef] = []
|
|
seen: set[str] = set()
|
|
for item in artifacts:
|
|
if not isinstance(item, dict):
|
|
return "[Error] every artifact must be an object with path and optional label"
|
|
try:
|
|
raw_path = str(item.get("path") or "")
|
|
if len(raw_path) > 1000:
|
|
return "[Error] artifact path is too long"
|
|
label = str(item.get("label") or "").strip()
|
|
if len(label) > 120:
|
|
return "[Error] artifact label is too long"
|
|
_, rel = resolve_artifact_path(
|
|
raw_path,
|
|
working_dir=self.working_dir,
|
|
user_root=self.user_root,
|
|
require_file=True,
|
|
allow_legacy_user_relative=False,
|
|
)
|
|
except ArtifactPathError as exc:
|
|
return f"[Error] cannot publish artifact: {exc}"
|
|
if rel in seen:
|
|
continue
|
|
seen.add(rel)
|
|
refs.append(ArtifactRef(path=rel, label=label))
|
|
|
|
names = ", ".join(ref.label or ref.path for ref in refs)
|
|
return ToolExecutionResult(
|
|
content=f"[OK] published {len(refs)} artifact(s): {names}",
|
|
artifacts=tuple(refs),
|
|
)
|