diff --git a/CHANGELOG.md b/CHANGELOG.md
index d809afa..1ab565b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,7 +8,7 @@
 
 ## Unreleased
 
-- 新增专业软件 Job 中心：Agent 可提交、查询和停止 Windows Node 上的受控软件任务；用户可在右下角跨对话查看进度、收到完成通知，并回到原对话分析结果。
+- 新增专业软件 Job 中心：Agent 可将当前对话目录内的数据登记为稳定输入，提交、查询和停止 Windows Node 上的受控软件任务；用户可在右下角跨对话查看进度、收到完成通知，并回到原对话分析结果。
 
 ## 0.65.2 — 2026-08-13
 
diff --git a/DESIGN.md b/DESIGN.md
index 5e24c3c..99ec267 100644
--- a/DESIGN.md
+++ b/DESIGN.md
@@ -466,13 +466,13 @@ Node 通过 `Authorization: Bearer` 与 `X-Node-Id` 建立 `/v1/software-nodes/c
 
 第二阶段已增加 `software_jobs`（专业软件任务）账本与 `origin.plot@v1` 的 offer/accept 骨架。用户只能在本人 task 下以幂等键提交固定 schema；云端规范化请求并记录 SHA-256，按当前进程真实在线、能力匹配、健康且有空闲 slot 的 Node 创建短期 offer。Node 再次校验 schema、图形类型和输出格式，使用 write-through、flush 与原子 rename 先落本机任务目录，再回 `job_accept`；重复 job 只有 digest 一致才接受。过期或发送失败的 offer 回到队列，lease、Node 和 digest 不匹配的响应被拒绝。Node 接收后云端进入 `dispatched` 而非 `running`，并将 slot 降为 0；只有固定 Worker 真正启动后才进入 `origin_running`。
 
-第三阶段补齐输入下载与恢复状态协议：`input_id` 固定为用户已有 artifact UUID，提交时快照文件名、大小和 SHA-256，只允许 CSV/XLSX/JSON 且不超过 100 MiB。Node 以自身 Bearer 身份访问任务绑定的只读下载端点，流式写入本 job 的 `input/`，同时限制声明大小并校验 SHA-256，完成后原子 rename；不暴露工作区路径。Node 会原子读取/补报 `terminal.json`，断线后云端把活动任务标记 `disconnected` 并保留 Node/lease，重连按 job、lease、digest 恢复下载或幂等补报终态，不自动重派。
+第三阶段补齐输入下载与恢复状态协议：`input_id` 固定为 artifact UUID。已有 artifact 可直接提交；普通 task 文件先调用 `register_artifact(path)` 登记稳定身份，该动作不把文件发布为聊天交付物。提交时快照文件名、大小和 SHA-256，只允许 CSV/XLSX/JSON 且不超过 100 MiB。Node 以自身 Bearer 身份访问任务绑定的只读下载端点，流式写入本 job 的 `input/`，同时限制声明大小并校验 SHA-256，完成后原子 rename；不暴露工作区路径。Node 会原子读取/补报 `terminal.json`，断线后云端把活动任务标记 `disconnected` 并保留 Node/lease，重连按 job、lease、digest 恢复下载或幂等补报终态，不自动重派。
 
 第四阶段落地固定 Origin Worker：Node 仅从管理员安装的固定 Python 运行时启动随程序发布的 `worker.py`，参数只有本机 job 目录；请求不能指定脚本、解释器或文件路径。Worker 使用 `originpro` 生成 OPJU、PNG、SVG、PDF、plot spec 和 provenance，校验产物签名并原子写入终态；当前受控图形仅含 line、scatter、line_scatter 和双栏出版布局。进程内 pipeline 按 job 去重，并脱离单次 WebSocket 的取消令牌运行；连接中断只延迟状态/终态上报。Node 进程若在 Worker 启动后重启，则保守失败而不重复驱动 Origin，避免无法证明的双执行。
 
 第五阶段完成输出上传与发布：Node 只按固定 manifest ID 逐项流式 PUT，并携带 Node、lease、request digest 与内容摘要；云端重新绑定任务身份，不信 Node 提供的路径或媒体类型。文件先进入用户根下隐藏暂存区，固定文件名、单文件/总大小和 SHA-256 全部验证后，目录级原子移动到 `<working_dir>/origin/<job_id>/`，再登记平台 artifact UUID 并写成功终态。重复 PUT、complete 和重连均按摘要幂等；部分上传不可见，只有完整集合才能发布。
 
-第六阶段增加用户级 Job 中心与 Agent typed tools。`software_capability_list` 只暴露固定能力及当前在线空闲节点数，`software_job_submit/status/cancel` 在构造时绑定当前 user/task，模型不能跨用户或跨对话指定归属。右下角 Job 中心按用户聚合各对话任务，活动期短轮询、空闲期降频；终态变化通知用户，成功任务可回到原对话发起分析。取消采用协作协议：未派发任务直接终止，已派发任务先进入 `cancelling`，云端通过 WebSocket 发送并在心跳时重放 `job_cancel`，Node 杀死固定 Worker 进程树后回报 `cancelled`；终态写入仍由云端账本裁决。
+第六阶段增加用户级 Job 中心与 Agent typed tools。`software_capability_list` 只暴露固定能力及当前在线空闲节点数，`software_job_submit/status/cancel` 在构造时绑定当前 user/task，模型不能跨用户或跨对话指定归属。`software_job_submit` 的模型可见唯一入口为 `input_id + plot + output`；`register_artifact` 是普通文件获得输入身份的唯一入口，并明确返回 UUID。内部完整 `request` 形式仅保留执行层兼容，不进入工具 schema。右下角 Job 中心按用户聚合各对话任务，活动期短轮询、空闲期降频；终态变化通知用户，成功任务可回到原对话发起分析。取消采用协作协议：未派发任务直接终止，已派发任务先进入 `cancelling`，云端通过 WebSocket 发送并在心跳时重放 `job_cancel`，Node 杀死固定 Worker 进程树后回报 `cancelled`；终态写入仍由云端账本裁决。
 
 后续仍需实现 Token 轮换；不得以任意命令或脚本接口临时代替。当前 Job 中心采用轮询而非用户事件推送，单活与 offer 选择仍只覆盖单 Web 进程；生产启用多实例前必须增加 Redis/PG fencing 或固定路由到单一控制面实例。
 
diff --git a/PROGRESS.md b/PROGRESS.md
index 7e24bde..1fa66ef 100644
--- a/PROGRESS.md
+++ b/PROGRESS.md
@@ -22,6 +22,8 @@
 
 ### 2026-08-13
 
+- **08-13 / Unreleased / 专业软件输入 artifact 契约**:新增 `register_artifact`，将 task 内普通文件登记为稳定输入但不发布聊天交付卡片；`software_job_submit` 改为模型可见的完整 Origin schema，并强制使用 artifact UUID，已有 artifact 可直接复用，内部完整 request 仅保留兼容。同步校准当前三种图形类型与运行/设计文档；相关 51 项 unittest、Python 编译和 diff 检查通过，未连接或写入生产 DB。
+
 - **08-13 / Unreleased / 专业软件 Job 中心**:Agent 新增固定能力发现、提交、状态和取消工具，后端提供用户级跨对话任务列表与协作取消，Windows Node 可终止固定 Worker 进程树并幂等回报取消终态；Web 右下角展示活动/最近任务、进度、完成通知、停止与回到原对话分析入口。相关 Python 78 项、JavaScript 语法、Python 编译、.NET build 与 diff 检查通过；数据库保持在 0031，未连库、未执行 migration。
 
 - **08-13 / 0.65.2 / 用户消息结构化附件 + 对话内图片预览**:新增 0031 `messages.attachment_refs`，新客户端将附件作为结构化字段发送，数据库正文仅保留用户自然语言；后端按 task working_dir 校验路径并在内存模型上下文中补附件提示，旧客户端与历史正文标记继续兼容。用户消息即时态和历史态统一展示附件 chip，图片额外显示可点击缩略图；相关 Python 34 项、全部前端 Node 26 项、Python/JavaScript 语法、Alembic 单 head 与 diff 检查通过，未连接或迁移生产 DB。
diff --git a/RUN.md b/RUN.md
index 4d4376b..d8ba648 100644
--- a/RUN.md
+++ b/RUN.md
@@ -1092,7 +1092,7 @@ install-windows-node.bat
 
 若 Python 未加入 PATH，可把绝对路径作为第一个参数，例如 `install-windows-node.bat "C:\Python312\python.exe"`。默认解释器为 `%ProgramData%\Zcbot\WindowsNode\runtimes\origin\Scripts\python.exe`。如需使用其他受管解释器，设置机器级 `ZCBOT_ORIGIN_PYTHON` 为绝对 `python.exe` 路径后重启 Node。`node.json`、可恢复任务和 runtime 集中保存在 `%ProgramData%\Zcbot\WindowsNode\`，不会因替换程序目录而丢失。运行时固定依赖见发布目录的 `origin-worker/requirements.txt`；任务请求无权选择解释器、脚本或路径。当前 Worker 支持 CSV/XLSX/JSON 输入，`line`、`scatter`、`line_scatter` 与 OPJU/PNG/SVG/PDF 输出。成功产物由 Node 流式上传，全部校验通过后发布到任务工作目录 `origin/<job_id>/`；上传中断会在重连时幂等续传。
 
-Web 用户登录后，右下角 Job 中心会聚合本人最近任务。活动任务约 4 秒刷新一次，空闲时降为约 30 秒；停止已派发任务是协作取消，状态先显示“正在停止”，Node 在线时立即接收，断线后在下次连接或心跳时重放。Agent 可调用 `software_capability_list`、`software_job_submit`、`software_job_status` 和 `software_job_cancel`；提交工具只创建固定 schema 的持久任务，不会阻塞当前对话等待完成。
+Web 用户登录后，右下角 Job 中心会聚合本人最近任务。活动任务约 4 秒刷新一次，空闲时降为约 30 秒；停止已派发任务是协作取消，状态先显示“正在停止”，Node 在线时立即接收，断线后在下次连接或心跳时重放。Agent 可调用 `software_capability_list`、`register_artifact`、`software_job_submit`、`software_job_status` 和 `software_job_cancel`。Origin 输入必须是 artifact：已有 UUID 可直接提交，普通 task 文件先用相对路径登记；登记不会发布聊天交付卡片。提交工具接收 `input_id`、`plot`、`output`，只创建固定 schema 的持久任务，不会阻塞当前对话等待完成。
 
 注册配置写入 `%ProgramData%\Zcbot\WindowsNode\node.json`；Token 使用 DPAPI `LocalMachine` 加密，ACL 仅允许注册账号和 `SYSTEM`。应始终用同一专用 Windows 账号执行统一安装器、注册并运行 Node。当前 MVP 以该账号的登录后计划任务启动，不安装 Windows Service。
 
diff --git a/core/artifact_lifecycle.py b/core/artifact_lifecycle.py
index b61b335..40ba821 100644
--- a/core/artifact_lifecycle.py
+++ b/core/artifact_lifecycle.py
@@ -45,6 +45,7 @@ def register_published_artifacts(
     user_root: Path,
     working_dir: Path,
     refs: tuple[dict, ...],
+    preserve_existing_labels: bool = False,
 ) -> tuple[dict, ...]:
     """Upsert active artifact identities and return version-2 message refs."""
     root = Path(user_root).resolve()
@@ -62,6 +63,14 @@ def register_published_artifacts(
             media_type = str(ref.get("media_type") or "") or mimetypes.guess_type(path.name)[0]
             size_bytes = path.stat().st_size
             content_sha256 = _hash_file(path)
+            update_values = {
+                "media_type": media_type,
+                "size_bytes": size_bytes,
+                "content_sha256": content_sha256,
+                "updated_at": datetime.now(timezone.utc),
+            }
+            if not preserve_existing_labels:
+                update_values["label"] = label
             statement = pg_insert(Artifact).values(
                 user_id=user_id,
                 origin_task_id=task_id,
@@ -73,13 +82,7 @@ def register_published_artifacts(
             ).on_conflict_do_update(
                 index_elements=[Artifact.user_id, Artifact.current_path],
                 index_where=Artifact.status == "active",
-                set_={
-                    "label": label,
-                    "media_type": media_type,
-                    "size_bytes": size_bytes,
-                    "content_sha256": content_sha256,
-                    "updated_at": datetime.now(timezone.utc),
-                },
+                set_=update_values,
             ).returning(Artifact.artifact_id)
             artifact_id = session.execute(statement).scalar_one()
             output.append(ArtifactRef(
@@ -91,6 +94,34 @@ def register_published_artifacts(
     return tuple(output)
 
 
+def register_workspace_artifact(
+    *,
+    user_id: UUID,
+    task_id: UUID,
+    user_root: Path,
+    working_dir: Path,
+    path: str,
+    label: str = "",
+) -> dict:
+    """Register one task-relative file for an internal platform workflow.
+
+    Unlike ``publish_artifacts``, this only creates or refreshes the stable artifact
+    identity. It does not attach the file to an assistant message or present it as a
+    final deliverable.
+    """
+    refs = register_published_artifacts(
+        user_id=user_id,
+        task_id=task_id,
+        user_root=user_root,
+        working_dir=working_dir,
+        refs=({"path": path, "label": label},),
+        preserve_existing_labels=True,
+    )
+    if len(refs) != 1 or not refs[0].get("artifact_id"):
+        raise ValueError("workspace artifact could not be registered")
+    return refs[0]
+
+
 def rename_active_artifacts(
     *,
     user_id: UUID,
diff --git a/core/tool_registry.py b/core/tool_registry.py
index 82f2f66..05f51c1 100644
--- a/core/tool_registry.py
+++ b/core/tool_registry.py
@@ -42,6 +42,7 @@ from tools.materials_project import (
 )
 from tools.office_to_pdf import OfficeToPdfTool
 from tools.read_document import ReadDocumentTool
+from tools.register_artifact import RegisterArtifactTool
 from tools.rename_working_dir import RenameWorkingDirTool
 from tools.run_python import RunPythonTool
 from tools.schedule import (
@@ -145,6 +146,12 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]:
                 working_dir=ctx.working_dir_path,
                 **wd_base,
             ),
+            RegisterArtifactTool(
+                ctx.uid,
+                ctx.task_id,
+                working_dir=ctx.working_dir_path,
+                **wd_base,
+            ),
         ]
 
     def _document_search() -> list:
diff --git a/docs/windows-node-design.md b/docs/windows-node-design.md
index 19b88ea..a8b0886 100644
--- a/docs/windows-node-design.md
+++ b/docs/windows-node-design.md
@@ -440,7 +440,7 @@ C# COM 只作未覆盖能力或旧版本备用，并与实际 Origin 版本匹
 ```json
 {
   "schema_version": 1,
-  "input": {"input_id": "data-01", "sheet": "Sheet1"},
+  "input": {"input_id": "5af18255-c859-43f3-a424-220bc68b7db2", "sheet": "Sheet1"},
   "plot": {
     "type": "line_scatter",
     "x": "Temperature",
@@ -461,7 +461,13 @@ C# COM 只作未覆盖能力或旧版本备用，并与实际 Origin 版本匹
 }
 ```
 
-首期图形类型：line、scatter、line_scatter、grouped_bar、box、histogram、heatmap。后续增加误差棒组合、三元图、等高线、三维曲面、XRD 堆叠图、热分析联图和多面板布局。
+当前图形类型：line、scatter、line_scatter。后续增加 grouped_bar、box、histogram、heatmap、误差棒组合、三元图、等高线、三维曲面、XRD 堆叠图、热分析联图和多面板布局。
+
+上述 JSON 同时是提交工具与云端账本、Windows Node 之间的 artifact 输入契约。
+Agent 优先复用已有 artifact UUID；普通 task 文件先调用
+`register_artifact(path)` 获得 UUID，再以 `input_id`、`plot`、`output` 调用
+`software_job_submit`。登记输入不会把文件发布为聊天交付物。旧的完整 `request`
+参数仅作为执行层兼容入口保留，不暴露在模型可见 schema 中。
 
 ### 12.3 模板与产物
 
diff --git a/tests/test_artifacts.py b/tests/test_artifacts.py
index 13d349e..dcce2cb 100644
--- a/tests/test_artifacts.py
+++ b/tests/test_artifacts.py
@@ -1,3 +1,4 @@
+import json
 import tempfile
 import unittest
 from contextlib import contextmanager
@@ -16,6 +17,7 @@ from core.artifacts import (
 from core.executor import ExecCtx
 from core.executor_host import HostExecutor
 from tools.publish_artifacts import PublishArtifactsTool
+from tools.register_artifact import RegisterArtifactTool
 from web.routers.files import _task_file_target
 
 
@@ -133,6 +135,43 @@ class ArtifactPathTests(unittest.TestCase):
         self.assertEqual(executed.content, "[OK] published 1 artifact(s): report.pdf")
         self.assertEqual(executed.artifacts[0]["path"], "report.pdf")
 
+    def test_register_artifact_returns_identity_without_publishing(self) -> None:
+        artifact_id = uuid4()
+        tool = RegisterArtifactTool(
+            uuid4(),
+            uuid4(),
+            working_dir=self.wd,
+            base_dir=self.wd,
+            user_root=self.root,
+        )
+        with patch(
+            "tools.register_artifact.register_workspace_artifact",
+            return_value={
+                "version": 2,
+                "artifact_id": str(artifact_id),
+                "scope": "working_dir",
+                "path": "report.pdf",
+            },
+        ) as register:
+            result = json.loads(tool.execute("report.pdf"))
+
+        self.assertEqual(result["artifact_id"], str(artifact_id))
+        self.assertEqual(register.call_args.kwargs["path"], "report.pdf")
+
+    def test_register_artifact_rejects_path_escape(self) -> None:
+        tool = RegisterArtifactTool(
+            uuid4(),
+            uuid4(),
+            working_dir=self.wd,
+            base_dir=self.wd,
+            user_root=self.root,
+        )
+        self.assertIn("cannot register artifact", tool.execute("../outside.csv"))
+
+    def test_register_artifact_schema_is_distinct_from_publish(self) -> None:
+        self.assertEqual(RegisterArtifactTool.parameters["required"], ["path"])
+        self.assertNotIn("artifacts", RegisterArtifactTool.parameters["properties"])
+
     def test_publish_path_is_strictly_task_relative_when_names_repeat(self) -> None:
         nested = self.wd / "技术讨论" / "nested.html"
         nested.parent.mkdir()
diff --git a/tests/test_software_job_tools.py b/tests/test_software_job_tools.py
index e5fa4d4..ad0523e 100644
--- a/tests/test_software_job_tools.py
+++ b/tests/test_software_job_tools.py
@@ -39,12 +39,42 @@ class SoftwareJobToolTests(unittest.TestCase):
 
     def test_submit_injects_current_user_and_task(self):
         created = {"job_id": str(uuid4()), "status": "queued"}
+        artifact_id = uuid4()
         tool = SoftwareJobSubmitTool(self.user_id, self.task_id)
         with patch("tools.software_jobs.create_job", return_value=(created, True)) as create:
-            result = json.loads(tool.execute("origin.plot@v1", {"input_id": str(uuid4())}))
+            result = json.loads(tool.execute(
+                "origin.plot@v1",
+                input_id=str(artifact_id),
+                plot={"type": "scatter", "x": "x", "y": ["y"]},
+                output={"formats": ["opju", "png"], "dpi": 300},
+            ))
         self.assertTrue(result["created"])
         self.assertEqual(create.call_args.args[:2], (self.user_id, self.task_id))
         self.assertEqual(create.call_args.kwargs["capability"], "origin.plot@v1")
+        self.assertEqual(
+            create.call_args.kwargs["request"],
+            {
+                "schema_version": 1,
+                "input": {"input_id": str(artifact_id)},
+                "plot": {"type": "scatter", "x": "x", "y": ["y"]},
+                "output": {"formats": ["opju", "png"], "dpi": 300},
+            },
+        )
+
+    def test_submit_requires_artifact_uuid(self):
+        result = SoftwareJobSubmitTool(self.user_id, self.task_id).execute(
+            "origin.plot@v1",
+            input_id="data/input.csv",
+            plot={"type": "scatter", "x": "x", "y": ["y"]},
+            output={"formats": ["png"]},
+        )
+        self.assertIn("call register_artifact first", result)
+
+    def test_submit_schema_requires_artifact_input(self):
+        required = SoftwareJobSubmitTool.parameters["required"]
+        self.assertIn("input_id", required)
+        self.assertNotIn("input_path", SoftwareJobSubmitTool.parameters["properties"])
+        self.assertNotIn("request", SoftwareJobSubmitTool.parameters["properties"])
 
     def test_status_and_cancel_reject_cross_task_job(self):
         foreign = {"job_id": str(uuid4()), "task_id": str(uuid4())}
diff --git a/tools/register_artifact.py b/tools/register_artifact.py
new file mode 100644
index 0000000..73067aa
--- /dev/null
+++ b/tools/register_artifact.py
@@ -0,0 +1,70 @@
+"""Register a task file as a stable artifact without publishing it to chat."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from uuid import UUID
+
+from core.artifact_lifecycle import register_workspace_artifact
+from core.artifacts import ArtifactPathError, resolve_artifact_path
+
+from .base import Tool
+
+
+class RegisterArtifactTool(Tool):
+    name = "register_artifact"
+    description = (
+        "Register one existing file in the current task as a stable artifact and return "
+        "its artifact_id. Use this before a tool that requires an artifact input. "
+        "Registration does not publish the file as a final chat deliverable."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "path": {
+                "type": "string",
+                "minLength": 1,
+                "maxLength": 1000,
+                "description": "File path relative to the current task working directory.",
+            }
+        },
+        "required": ["path"],
+        "additionalProperties": False,
+    }
+
+    def __init__(
+        self,
+        user_id: UUID,
+        task_id: UUID,
+        *,
+        working_dir: Path,
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self.user_id = user_id
+        self.task_id = task_id
+        self.working_dir = Path(working_dir)
+
+    def execute(self, path: str) -> str:
+        if self.user_root is None:
+            return "[Error] artifact registration requires a user workspace"
+        if len(str(path or "")) > 1000:
+            return "[Error] artifact path is too long"
+        try:
+            _, relative_path = resolve_artifact_path(
+                path,
+                working_dir=self.working_dir,
+                user_root=self.user_root,
+                require_file=True,
+                allow_legacy_user_relative=False,
+            )
+            artifact = register_workspace_artifact(
+                user_id=self.user_id,
+                task_id=self.task_id,
+                user_root=self.user_root,
+                working_dir=self.working_dir,
+                path=relative_path,
+            )
+        except (ArtifactPathError, ValueError) as exc:
+            return f"[Error] cannot register artifact: {exc}"
+        return json.dumps(artifact, ensure_ascii=False)
diff --git a/tools/software_jobs.py b/tools/software_jobs.py
index c8b13fa..57c05ef 100644
--- a/tools/software_jobs.py
+++ b/tools/software_jobs.py
@@ -46,34 +46,138 @@ class SoftwareCapabilityListTool(_SoftwareJobTool):
 class SoftwareJobSubmitTool(_SoftwareJobTool):
     name = "software_job_submit"
     description = (
-        "Submit a persistent professional software job for the current task. "
+        "Submit an Origin plot job using a registered CSV, XLSX, or JSON artifact. "
+        "Call register_artifact first when the input is only a workspace file. "
         "Return immediately with job_id; do not poll continuously or wait for completion."
     )
+    _axis_schema = {
+        "type": "object",
+        "properties": {
+            "title": {"type": "string"},
+            "unit": {"type": "string"},
+            "scale": {"type": "string", "enum": ["linear"]},
+        },
+        "additionalProperties": False,
+    }
+    _plot_schema = {
+        "type": "object",
+        "properties": {
+            "type": {"type": "string", "enum": ["line", "scatter", "line_scatter"]},
+            "x": {
+                "type": "string",
+                "minLength": 1,
+                "maxLength": 128,
+                "description": "Exact X column name in the input file.",
+            },
+            "y": {
+                "type": "array",
+                "minItems": 1,
+                "maxItems": 16,
+                "uniqueItems": True,
+                "items": {"type": "string", "minLength": 1, "maxLength": 128},
+                "description": "One to sixteen unique Y column names.",
+            },
+            "template": {"type": "string", "enum": ["publication_double_column"]},
+            "title": {"type": "string", "maxLength": 500},
+            "x_axis": _axis_schema,
+            "y_axis": _axis_schema,
+            "legend": {
+                "type": "object",
+                "properties": {
+                    "enabled": {"type": "boolean", "enum": [True]},
+                    "position": {"type": "string", "enum": ["top_right"]},
+                },
+                "additionalProperties": False,
+            },
+            "error_bars": {"type": "null"},
+        },
+        "required": ["type", "x", "y"],
+        "additionalProperties": False,
+    }
+    _output_schema = {
+        "type": "object",
+        "properties": {
+            "formats": {
+                "type": "array",
+                "minItems": 1,
+                "uniqueItems": True,
+                "items": {"type": "string", "enum": ["opju", "png", "svg", "pdf"]},
+            },
+            "dpi": {"type": "integer", "minimum": 72, "maximum": 1200},
+            "capture_screenshots": {"type": "boolean"},
+            "record_video": {"type": "boolean", "enum": [False]},
+        },
+        "required": ["formats"],
+        "additionalProperties": False,
+    }
     parameters = {
         "type": "object",
         "properties": {
             "capability": {"type": "string", "enum": sorted(SUPPORTED_CAPABILITIES)},
-            "request": {"type": "object"},
+            "input_id": {
+                "type": "string",
+                "description": "Artifact UUID returned by register_artifact or another artifact-producing flow.",
+            },
+            "sheet": {
+                "type": "string",
+                "minLength": 1,
+                "maxLength": 128,
+                "description": "Optional XLSX worksheet name.",
+            },
+            "plot": _plot_schema,
+            "output": _output_schema,
             "idempotency_key": {
                 "type": "string",
                 "description": "Stable unique key for this exact submission; omit to generate one.",
             },
         },
-        "required": ["capability", "request"],
+        "required": ["capability", "input_id", "plot", "output"],
         "additionalProperties": False,
     }
 
-    def execute(self, capability: str, request: dict, idempotency_key: str = "") -> str:
+    def execute(
+        self,
+        capability: str,
+        input_id: str = "",
+        plot: dict | None = None,
+        output: dict | None = None,
+        sheet: str = "",
+        request: dict | None = None,
+        idempotency_key: str = "",
+    ) -> str:
         try:
+            if isinstance(input_id, dict) and request is None:
+                request = input_id
+                input_id = ""
+            if request is not None:
+                if input_id or plot is not None or output is not None or sheet:
+                    return "[Error] request cannot be combined with input_id, sheet, plot, or output"
+                normalized_request = request
+            else:
+                if not input_id or plot is None or output is None:
+                    return "[Error] input_id, plot, and output are required for an Origin job"
+                try:
+                    canonical_input_id = str(UUID(str(input_id)))
+                except ValueError:
+                    return "[Error] input_id must be an artifact UUID; call register_artifact first"
+                input_spec = {"input_id": canonical_input_id}
+                if sheet:
+                    input_spec["sheet"] = sheet
+                normalized_request = {
+                    "schema_version": 1,
+                    "input": input_spec,
+                    "plot": plot,
+                    "output": output,
+                }
             job, created = create_job(
                 self.user_id,
                 self.task_id,
                 idempotency_key=idempotency_key.strip() or str(uuid4()),
                 capability=capability,
-                request=request,
+                request=normalized_request,
             )
             return json.dumps({**job, "created": created}, ensure_ascii=False)
-        except SoftwareJobError as exc:
+        except (SoftwareJobError, ValueError) as exc:
             return f"[Error] {exc}"