From ebde18169566025a1a3453c8b0978816d18a008f Mon Sep 17 00:00:00 2001 From: Cyrus Zhang Date: Fri, 16 Oct 2026 11:34:43 -0700 Subject: [PATCH 1/2] feat(engine): integrate durable PawApp task execution Signed-off-by: Cyrus Zhang --- CHANGELOG.md | 14 + docs/design/pawapp-submission-protocol.md | 290 ++++++ packages/qwenpaw-data-host-core/README.md | 11 +- .../host/core/agent/host_capabilities.py | 378 ++++++++ .../host/core/agent/task_experience.py | 50 ++ .../qwenpaw_data/host/core/agent/toolkit.py | 110 ++- .../src/qwenpaw_data/host/core/api/app.py | 20 + .../src/qwenpaw_data/host/core/api/deps.py | 8 +- .../host/core/api/models/artifact.py | 22 + .../host/core/api/models/runtime_models.py | 6 +- .../host/core/api/models/stream_objects.py | 8 + .../host/core/api/models/submissions.py | 156 ++++ .../host/core/api/routers/artifacts.py | 67 +- .../host/core/api/routers/chats.py | 19 + .../host/core/api/routers/submissions.py | 385 ++++++++ .../src/qwenpaw_data/host/core/core.py | 13 +- .../src/qwenpaw_data/host/core/db/tables.py | 36 +- .../host/core/runtime/chat_runtime.py | 135 ++- .../qwenpaw_data/host/core/store/sql_store.py | 86 +- .../host/core/store/submissions.py | 370 ++++++++ .../tests/test_artifact_registration.py | 83 ++ .../tests/test_artifacts_api.py | 23 + .../tests/test_host_capabilities.py | 231 +++++ .../tests/test_host_runtime.py | 31 + .../tests/test_preferences_api.py | 47 + .../tests/test_runtime_model_bindings.py | 2 + .../tests/test_stream_objects.py | 3 + .../tests/test_submissions_api.py | 841 ++++++++++++++++++ .../tests/test_task_experience_tools.py | 25 + 29 files changed, 3373 insertions(+), 97 deletions(-) create mode 100644 docs/design/pawapp-submission-protocol.md create mode 100644 packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/host_capabilities.py create mode 100644 packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/task_experience.py create mode 100644 packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/submissions.py create mode 100644 packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/submissions.py create mode 100644 packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/store/submissions.py create mode 100644 packages/qwenpaw-data-host-core/tests/test_artifact_registration.py create mode 100644 packages/qwenpaw-data-host-core/tests/test_host_capabilities.py create mode 100644 packages/qwenpaw-data-host-core/tests/test_submissions_api.py create mode 100644 packages/qwenpaw-data-host-core/tests/test_task_experience_tools.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 795a660..b419714 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,20 @@ follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Added + +- Durable Engine submissions for PawApp backends: atomic session/run creation, + idempotent retries, lookup by submission ID, and replayable events. SQL startup + recovery preserves accepted run identities and reports interrupted execution + without automatically rerunning it. See the + [submission protocol](docs/design/pawapp-submission-protocol.md); JSON mode + explicitly declines this protocol. +- Durable submission commands let a PawApp backend answer the active + clarification or cancel the original run with idempotent command receipts, + scoped lookup, conflict detection, and explicit unknown recovery state. +- Task-scoped Host tools and skills, verified artifact handoff with digest-bound + reads, and explicit analysis progress and deliverable presentation metadata. + ## [0.3.2] - Unreleased ### Added diff --git a/docs/design/pawapp-submission-protocol.md b/docs/design/pawapp-submission-protocol.md new file mode 100644 index 0000000..665e8ef --- /dev/null +++ b/docs/design/pawapp-submission-protocol.md @@ -0,0 +1,290 @@ +# PawApp vNext: durable Engine submissions + +This contract lets a trusted PawApp backend start an independent analysis +without a Main Chat turn or an existing Engine session. The Engine owns +execution and its replay log; the QwenPaw Host owns task identity, scope checks, +Direct/Delegated presentation, and any Main Agent continuation. + +## Release boundary after runtime 0.3.2 + +This branch is based on `14535e3` (QwenPaw-Data #81). Runtime 0.3.2 already +provides the native console API, per-turn model bindings, datasource/dataset +selection, scoped session ownership, readable session names, and the Windows +artifact-path fixes. Those features and the release dependency updates are +inherited from `main`, rather than reintroduced by this branch. + +The remaining changes add durable submissions and commands, task-scoped Host +capabilities, verified artifact handoff, and analysis progress/presentation. +The native console continues to use its existing API; a Host adapter opts into +the durable protocol through capability discovery. Package versions remain at +the 0.3.2 base until a separate release is prepared. + +## Capability discovery + +This branch implements **submission protocol 1**. It does not declare a released +Engine version compatible with all of PawApp vNext. Adapters must probe +`GET /api/v1/capabilities/submissions` and require `protocol_version: 1`, +`durable_submissions: true`, and `event_replay: true`. Task answer/cancel also +requires `durable_commands: true`. Older Engines lack this +endpoint; JSON mode reports unsupported and returns 501 for submit/query/replay. +Do not silently fall back to non-idempotent session/chat creation. + +Engines that accept task-scoped Host tools and Skills additionally advertise +`scoped_host_capabilities: true`. A Host that supplies a capability envelope +must require this flag before submission; older Engines remain compatible with +submissions that do not carry an envelope. + +Engines used by an artifact-aware adapter advertise `artifact_handoff: true`. +After every successful tool result they emit `artifact.registered` for changed +files even when BizTrace is disabled. Each event includes normalized path, +media type, byte size, and `sha256:` digest. A Host copies the file through +`GET /api/v1/sessions/{session_id}/artifacts/file?path=...&digest=...`; the +Engine requires the caller to own the session and returns 409 if the live file +no longer matches that immutable event version. + +`GET /api/v1/capabilities/analysis` returns readiness version 1 and the boolean +`model_configured`. It checks the explicit model, or the same local Agent +Configuration preferences/environment fallback used by independent runs. +With `local_only=true`, it instead checks the trusted caller's saved default +model. No credentials +or configuration error details are returned, and no provider call is made. +This is configuration readiness, not a connectivity guarantee. The Host also +checks its authorized datasource against `/api/v1/datasources`, then returns a +blocked result with the App settings entry before creating a task if setup is +missing. Durable submission protocol support alone does not imply readiness. + +## Submit and reconcile + +### Additive capabilities: analysis experience and existing sessions + +`analysis_experience: true` advertises typed `analysis.progress` events with +stage `read_data`, `confirm_scope`, `analyze`, or `publish_report`. Private +runtime tools let the domain agent declare stages and publish selected files. +Ordinary artifact discovery emits explicit `diagnostic` / `app_only` metadata; +filenames are not a publication decision. Intentional artifact publication +verifies the file and adds the shared presentation fields: `schema_version: 1`, +`role`, `kind`, `visibility`, `preview`, and `rank`. Host owns generic rendering +and authorization; Engine owns the domain meaning and selected deliverables. + +`session_submissions: true` permits optional `session_id`, `attachment_ids`, +and `artifact_comments` in the submission below. The existing session must +belong to the supplied trusted namespace. Attachments must belong to the same +session and namespace. A running turn blocks a second new submission; retries +of the original receipt remain idempotent. Session update, chat creation and +receipt commit in one transaction. Subsequent completed turns may reuse the +same session. All context participates in the digest; omitted/empty additive +fields retain the original protocol-1 digest for old accepted submissions. +`GET /api/v1/sessions/{session_id}/submission-readiness` checks ownership and +returns `ready: false` while a turn is running; unavailable/foreign sessions +return 404. This read-only preflight prevents admission of legacy or busy +sessions; the transactional submission still rechecks both conditions. + +The vNext Host adapter creates native sessions under the same scoped +namespace, admits each turn as a Direct task, and routes answers/cancellation +through durable commands. It must not forward browser raw submissions. Old +unscoped sessions require an explicit migration or a new session, not implicit +ownership reassignment. Dataset selection is supported when `dataset_context` +is advertised: `dataset_names` is validated, persisted on the chat, and included +in the request digest and agent input. New sessions require a datasource; +existing sessions may omit it when `optional_session_datasource` is advertised. +Disabling clarification is not supported by this submission contract and must +not be silently ignored. + +```http +POST /api/v1/submissions +Authorization: Bearer +X-User-Id: +Content-Type: application/json + +{ + "protocol_version": 1, + "submission_id": "sub_host_generated_id", + "text": "Analyze revenue for the selected datasource", + "datasource_id": "ds_revenue", + "agent_id": "default", + "capability_bridge": { + "protocol_version": 1, + "endpoint": "http://127.0.0.1:8088/api/pawapp-capabilities", + "token": "" + } +} +``` + +`capability_bridge` is optional and ephemeral. The Engine keeps it in the run's +in-memory request context, uses it only for the Host capability protocol, and +never writes it to workspace files or event streams. The endpoint must be an +absolute HTTP(S) URL without credentials, query, or fragment. The token is +included in the idempotency digest, so a Host retry must reproduce the same +envelope for a given durable submission. + +`protocol_version` and `agent_id` default to 1 and `default`. The ID accepts +1–128 ASCII letters, digits, underscores, or hyphens. `text` and `datasource_id` +must be nonblank; unknown fields and unsupported protocol versions are rejected. +Datasource existence, model readiness, and analysis execution can still fail +after acceptance. Acceptance is not a readiness guarantee. + +Successful acceptance and identical retries both return **202**: + +```json +{ + "protocol_version": 1, + "submission_id": "sub_host_generated_id", + "state": "accepted", + "run": { + "session_id": "ses_generated_id", + "run_id": "chat_generated_id", + "status": "running", + "last_sequence_number": -1, + "error": null, + "reason": null + } +} +``` + +The `submissions` SQL table has a unique `(user_id, submission_id)` key and +stores a request digest plus immutable session/run IDs. The digest covers the +effective protocol version, agent, text, datasource, and optional capability +envelope. The receipt, new +session, and first chat commit in **one transaction**. Only the transaction +winner schedules the runtime; concurrent identical retries return its mapping. +Reusing an ID with different inputs returns **409**. Different submission IDs +create different sessions, allowing concurrent delegations to the same App. + +HTTP request cancellation does not cancel the acceptance/scheduling operation. +A process exit after commit but before scheduling leaves an accepted run that +startup recovery marks interrupted; it never schedules a replacement run. +This is deduplicated acceptance, not an exactly-once guarantee for external +tool side effects. + +`GET /api/v1/submissions/{submission_id}` returns: + +| Outcome | Meaning for the Host adapter | +|---|---| +| 200, `state: accepted` | Bind the original session/run IDs; attach using the last cursor committed by Host. | +| 200, `state: not_found`, `run: null` | No receipt exists in this caller namespace. Retry with the **same** ID and inputs only if Host has never bound a run. | +| Timeout, 5xx, invalid response, unavailable Engine | Submission outcome is **unknown**. Keep the original ID and reconcile later; never reinterpret transport failure as `not_found`. | +| Accepted with `run.status: unknown`, `reason: run_missing` | Receipt remains, but its run is unavailable. Preserve the mapping and report unresolved; do not create a replacement. | + +Receipts have no TTL and are not deleted when a session is soft-deleted. If a +run is missing, replay returns 410 and retries still return the original receipt. +Durability assumes the same persistent database: deleting/restoring/replacing +that database is not a supported transparent failover operation. + +## Durable answer and cancel commands + +`POST /api/v1/submissions/{submission_id}/commands` applies a command to the +original accepted run. The request contains `protocol_version: 1`, a stable +`command_id`, and one of these payloads: + +```json +{ + "command_id": "answer_clarification_1", + "kind": "answer", + "request_id": "clarification_1", + "answers": [{ + "question": "Which period?", + "selected_options": ["Q1"], + "custom_text": null + }] +} +``` + +```json +{ + "command_id": "cancel_task_1", + "kind": "cancel", + "reason": "No longer needed" +} +``` + +The `submission_commands` table permanently binds +`(user_id, submission_id, command_id)` to a kind and request digest before the +runtime is touched. Identical retries return the stored receipt and never apply +the effect twice. Reusing a command ID with different content returns 409. +Commands and their lookups use the submission's caller namespace, so a receipt +cannot be read or applied through another namespace. + +The response state is `accepted`, `rejected`, or `unknown`. An answer is rejected +with `reason: stale_request` when its clarification is absent, already resolved, +or no longer belongs to the active runtime. Cancel targets the bound run; a +terminal run returns an accepted receipt with `reason: already_terminal` and +retains its output. A crash after the command receipt is prepared but before an +outcome is committed is reported as `unknown`; it is never guessed from a lost +HTTP response. + +`GET /api/v1/submissions/{submission_id}/commands/{command_id}` returns the same +receipt, `unknown` for a prepared command, or `not_found` when no receipt exists. +The Host must persist its own command before POST, reconcile an uncertain send by +this lookup, and keep the same command ID on every retry. + +## Replay and recovery + +`GET /api/v1/submissions/{submission_id}/events` resolves the receipt under the +same caller namespace and reuses the existing chat SSE stream. `run_id` is the +Engine's `chat_id`; it is not a second execution identity. Every persisted frame +contains session/chat IDs and a dense, monotonic `sequence_number` (starting at +0); its SSE `id` is the cursor. Pass `Last-Event-ID` or `after_sequence_number` +on reconnect; when both are present, the larger cursor wins. + +Host must persist its task→session/run mapping and consumed cursor with its own +task/event update. `last_sequence_number` in lookup is an Engine watermark, +**not** permission to skip events Host has not consumed. Reconnect can redeliver +events; consumers deduplicate by run identity and sequence. + +| Failure | Result | +|---|---| +| Host disconnect/reconnect, Engine still running | Replay from Host's committed cursor, then follow live events; no new analysis. | +| Engine restart with a persisted terminal response | Preserve its outcome and replay the same terminal event. | +| Engine restart with an unfinished accepted run | Atomically persist cancellation plus an error with `details.reason: executor_restarted`; submission status is `interrupted`. No automatic rerun. | +| Chat row says `completed`, but no terminal response exists | While running, lookup says `reconciling`; after restart, `interrupted`. The runtime may have lost final message data before flushing it. Never synthesize success. | +| Persisted failed/canceled chat lacks its final response | Startup appends the corresponding terminal event without executing again. | + +For wire compatibility, interrupted runs use a `response` event with +`status: cancelled`, `error.code: VALIDATION`, and +`error.details.reason: executor_restarted`. The adapter maps that reason to +Host `interrupted`, separately from user cancellation. Lookup statuses are +`running`, `reconciling`, `succeeded`, `failed`, `cancelled`, `interrupted`, or +`unknown`. A successful outcome requires a persisted `response/completed`; +**EOF is never success**. Result text remains in the existing message/content +events and must be reconstructed by the adapter, including full snapshots +versus deltas. + +Startup recovery runs before requests or scheduled jobs and commits chat state, +any missing terminal event, and its sequence watermark together. Recovery can +run repeatedly without appending duplicate terminal events. Existing console +and session/chat APIs retain their behavior; their POST endpoints do not gain +idempotency from this protocol. + +## Trust and deployment boundary + +Run one Engine process/worker per database. SQL unique keys protect concurrent +submissions, but the execution registry and live event fan-out remain in one +process; this is not a multi-worker scheduler. SQLite/WAL is the verified +backend. The SQL uses SQLAlchemy's portable expressions, but PostgreSQL recovery +has not been exercised in this change. + +All routes use the existing service bearer/loopback middleware. `X-User-Id` +partitions receipts as a **trusted backend identity stamp**, not an authenticated +end-user principal. A service token holder can choose that header. The Host +must derive a stable namespace from its trusted context, enforce user/workspace/ +App permissions, and keep the Engine token out of the browser. This change +does not add tenant isolation to older session/chat endpoints. + +## Validation and integration boundary + +The API tests use real SQLite transactions and HTTP routes with a controlled +runtime; no model/provider calls are needed. They cover concurrent duplicate +submissions, input conflicts, separate runs and caller namespaces, transaction +rollback, lost responses, cancellation of HTTP requests, hard process exit +after commit, restart recovery, live disconnect/replay, retained receipts after +history deletion, unavailable databases, bearer checks, and unsupported JSON +mode. Command tests cover idempotent answer/cancel effects, content conflicts, +stale clarification receipts, namespace isolation, and the prepared/unknown +recovery window. Existing host-core regression tests cover console/chat +compatibility. + +The QwenPaw Host Data adapter consumes these submission and command routes and +projects replayed Engine output into Host TaskEvents. Main Agent continuation +with arbitrary follow-on tools, grant UI, and public/private skill/tool access +remain separate implementation gates. The protocol probe above must not be used +to claim those gates are complete. diff --git a/packages/qwenpaw-data-host-core/README.md b/packages/qwenpaw-data-host-core/README.md index 89a6659..ae9f16b 100644 --- a/packages/qwenpaw-data-host-core/README.md +++ b/packages/qwenpaw-data-host-core/README.md @@ -32,6 +32,10 @@ worker. Without `QWENPAW_DATA_API_TOKEN`, only loopback clients are accepted. |---|---| | `POST /sessions` · `GET /sessions` · `GET/PATCH/DELETE /sessions/{id}` | Session lifecycle: create, list (search/filter/sort/pagination), rename, soft-delete | | `POST /sessions/{sid}/chats` | Start a turn in an existing session: `{"text": ..., "datasource_id": ...}` → `{"chat": {...}}`; 404 without a session, 409 while a turn is active | +| `GET /capabilities/submissions` | Discover durable submission protocol support (SQL only) | +| `POST /submissions` · `GET /submissions/{id}` | Atomically create an independent session/run with an idempotent submission ID; query its original identity and status | +| `POST /submissions/{id}/commands` · `GET /submissions/{id}/commands/{command_id}` | Durably answer the run's active clarification or cancel the original run; retry and reconcile by command ID | +| `GET /submissions/{id}/events` | Replay and follow the accepted run's persisted events with the same SSE cursor contract | | `GET /sessions/{sid}/chats` | List a session's chats | | `POST /sessions/{sid}/chats/{cid}/stop` | Cancel a running turn | | `GET /sessions/{sid}/chats/{cid}/events` | SSE stream: replays persisted history, then live events; supports `Last-Event-ID` (or `after_sequence_number`) resume | @@ -52,6 +56,9 @@ protocol with two interchangeable backends: file layout (`/host/{chats,sessions,preferences}/`). A conformance test suite runs every store test against both backends. +Durable submissions require SQL; JSON mode explicitly returns 501 for this +protocol. See the [submission contract](../../../docs/design/pawapp-submission-protocol.md) +for retry, recovery, authentication, and compatibility requirements. When no explicit model is passed, the service resolves the local user's configured active model from preferences and falls back to env vars. @@ -64,7 +71,9 @@ Each SSE frame carries `id` (dense per-chat sequence number), `event` close the stream) · `message` / `content` (assistant text, reasoning, tool calls and outputs, media) · `task_status` (DAG plan snapshots) · `biz_event` / `segment` / `artifact.registered` / `followup.generated` -(reserved; producers arrive in later waves). +(structured execution metadata). Artifact registrations include media type, +size, and SHA-256 digest. Digest-bound artifact reads verify session ownership +and reject content that changed after registration. ### Environment variables diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/host_capabilities.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/host_capabilities.py new file mode 100644 index 0000000..5a79f8a --- /dev/null +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/host_capabilities.py @@ -0,0 +1,378 @@ +# -*- coding: utf-8 -*- +"""Task-scoped tools and skills imported from the QwenPaw Host.""" + +from __future__ import annotations + +import asyncio +import base64 +import binascii +import json +import os +import shutil +import tempfile +import threading +from pathlib import Path, PurePosixPath +from typing import Any +from urllib.parse import quote + +import frontmatter +import httpx +from agentscope.message import TextBlock, ToolResultState +from agentscope.skill import Skill +from agentscope.tool import FunctionTool, ToolChunk + +_MAX_SKILL_BYTES = 2 * 1024 * 1024 +_MAX_SKILL_FILES = 1024 +_VALID_STATES = {item.value: item for item in ToolResultState} +_MATERIALIZE_LOCK = threading.Lock() + + +class HostCapabilityError(RuntimeError): + """A stable protocol or authorization failure from the Host bridge.""" + + def __init__(self, code: str): + self.code = code + super().__init__(code) + + +class HostCapabilityClient: + """Consume the narrow, task-bound capability protocol exposed by Host.""" + + def __init__( + self, + config: dict[str, Any], + *, + storage_workspace_dir: str | Path, + runtime_workspace_dir: str | Path, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: + if config.get("protocol_version") != 1: + raise HostCapabilityError("unsupported_host_capability_protocol") + endpoint = config.get("endpoint") + token = config.get("token") + if not isinstance(endpoint, str) or not isinstance(token, str): + raise HostCapabilityError("invalid_host_capability_bridge") + try: + url = httpx.URL(endpoint) + except (TypeError, ValueError, httpx.InvalidURL): + raise HostCapabilityError("invalid_host_capability_bridge") from None + if ( + url.scheme not in {"http", "https"} + or not url.host + or url.userinfo + or url.query + or url.fragment + or not token + ): + raise HostCapabilityError("invalid_host_capability_bridge") + self._storage_root = Path(storage_workspace_dir) / ".pawapp-capabilities" + self._runtime_root = Path(runtime_workspace_dir) / ".pawapp-capabilities" + self._client = httpx.AsyncClient( + base_url=str(url).rstrip("/") + "/", + headers={"Authorization": f"Bearer {token}"}, + transport=transport, + timeout=httpx.Timeout(60.0, connect=5.0), + follow_redirects=False, + trust_env=False, + ) + self._catalog: list[dict[str, Any]] | None = None + + async def aclose(self) -> None: + await self._client.aclose() + + async def catalog(self) -> list[dict[str, Any]]: + if self._catalog is not None: + return list(self._catalog) + payload = await self._request("GET", "catalog") + if payload.get("protocol_version") != 1: + raise HostCapabilityError("unsupported_host_capability_protocol") + entries = payload.get("capabilities") + if not isinstance(entries, list): + raise HostCapabilityError("invalid_host_capability_response") + parsed = [self._descriptor(item) for item in entries] + ids = [item["capability_id"] for item in parsed] + tool_names = [item["wire_name"] for item in parsed if item["kind"] == "tool"] + skill_names = [item["wire_name"] for item in parsed if item["kind"] == "skill"] + if ( + len(ids) != len(set(ids)) + or len(tool_names) != len(set(tool_names)) + or len(skill_names) != len(set(skill_names)) + ): + raise HostCapabilityError("invalid_host_capability_response") + self._catalog = parsed + return list(parsed) + + async def build_tools(self) -> list[FunctionTool]: + tools: list[FunctionTool] = [] + for descriptor in await self.catalog(): + if descriptor["kind"] != "tool": + continue + tools.append( + FunctionTool( + self._remote_tool(descriptor), + name=descriptor["wire_name"], + description=descriptor["description"], + input_schema=descriptor["input_schema"], + is_read_only=descriptor["is_read_only"], + ), + ) + return tools + + async def load_skills(self) -> list[Skill]: + skills: list[Skill] = [] + for descriptor in await self.catalog(): + if descriptor["kind"] != "skill": + continue + if not descriptor["available"]: + raise HostCapabilityError( + descriptor["blocked_reason"] or "skill_dependency_unavailable", + ) + capability_id = quote(descriptor["capability_id"], safe="") + payload = await self._request("GET", f"skills/{capability_id}") + if payload.get("protocol_version") != 1: + raise HostCapabilityError( + "unsupported_host_capability_protocol", + ) + files = payload.get("files") + returned_descriptor = self._descriptor(payload.get("descriptor")) + if returned_descriptor != descriptor: + raise HostCapabilityError("invalid_host_capability_response") + if not isinstance(files, list): + raise HostCapabilityError("invalid_host_capability_response") + directory = await self._materialize(descriptor, files) + try: + skill_text = (directory / "SKILL.md").read_text(encoding="utf-8") + content = frontmatter.loads(skill_text) + except (OSError, UnicodeDecodeError, ValueError): + raise HostCapabilityError("invalid_host_skill") from None + if not content.get("name") or not content.get("description"): + raise HostCapabilityError("invalid_host_skill") + runtime_dir = self._runtime_root / descriptor["digest"] + skills.append( + Skill( + name=descriptor["wire_name"], + description=descriptor["description"], + dir=str(runtime_dir), + markdown=content.content, + updated_at=(directory / "SKILL.md").stat().st_mtime, + ), + ) + return skills + + def _remote_tool(self, descriptor: dict[str, Any]): + async def invoke(**kwargs: Any) -> ToolChunk: + capability_id = quote(descriptor["capability_id"], safe="") + try: + payload = await self._request( + "POST", + f"tools/{capability_id}/invoke", + json={"params": kwargs}, + ) + except HostCapabilityError as exc: + denied = exc.code in { + "capability_not_found", + "invalid_capability_scope", + "invalid_capability_token", + "tool_permission_denied", + } + return ToolChunk( + content=[TextBlock(text=f"HostCapabilityError: {exc.code}")], + state=(ToolResultState.DENIED if denied else ToolResultState.ERROR), + ) + if ( + payload.get("protocol_version") != 1 + or payload.get("state") != "success" + ): + return ToolChunk( + content=[ + TextBlock( + text="HostCapabilityError: invalid_host_capability_response" + ), + ], + state=ToolResultState.ERROR, + ) + return self._as_tool_chunk(payload.get("output")) + + return invoke + + async def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]: + try: + response = await self._client.request(method, path, **kwargs) + except httpx.TransportError: + raise HostCapabilityError("host_capability_unavailable") from None + try: + payload = response.json() + except ValueError: + raise HostCapabilityError("invalid_host_capability_response") from None + if not isinstance(payload, dict): + raise HostCapabilityError("invalid_host_capability_response") + if not response.is_success: + detail = payload.get("detail") + code = detail if isinstance(detail, str) else "host_capability_rejected" + raise HostCapabilityError(code) + return payload + + @staticmethod + def _descriptor(value: Any) -> dict[str, Any]: + if not isinstance(value, dict): + raise HostCapabilityError("invalid_host_capability_response") + required_strings = ("capability_id", "name", "wire_name", "scope", "digest") + if any( + not isinstance(value.get(key), str) or not value[key] + for key in required_strings + ): + raise HostCapabilityError("invalid_host_capability_response") + kind = value.get("kind") + scope = value.get("scope") + if kind not in {"tool", "skill"}: + raise HostCapabilityError("invalid_host_capability_response") + if scope not in {"host_public", "app_private"}: + raise HostCapabilityError("invalid_host_capability_response") + expected_segment = "tool" if kind == "tool" else "skill" + if not value["capability_id"].startswith( + (f"host/{expected_segment}/", f"app/{expected_segment}/"), + ): + raise HostCapabilityError("invalid_host_capability_response") + description = value.get("description", "") + available = value.get("available", True) + blocked_reason = value.get("blocked_reason") + is_read_only = value.get("is_read_only", False) + if ( + not isinstance(description, str) + or not isinstance(available, bool) + or not isinstance(is_read_only, bool) + or (blocked_reason is not None and not isinstance(blocked_reason, str)) + ): + raise HostCapabilityError("invalid_host_capability_response") + schema = value.get("input_schema") + if kind == "tool" and not isinstance(schema, dict): + raise HostCapabilityError("invalid_host_capability_response") + digest = value["digest"] + if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest): + raise HostCapabilityError("invalid_host_capability_response") + return { + "capability_id": value["capability_id"], + "name": value["name"], + "wire_name": value["wire_name"], + "kind": kind, + "scope": scope, + "description": description, + "input_schema": schema, + "digest": digest, + "available": available, + "blocked_reason": blocked_reason, + "is_read_only": is_read_only, + } + + async def _materialize( + self, + descriptor: dict[str, Any], + files: list[Any], + ) -> Path: + return await asyncio.to_thread( + self._materialize_sync, + descriptor, + files, + ) + + def _materialize_sync( + self, + descriptor: dict[str, Any], + files: list[Any], + ) -> Path: + with _MATERIALIZE_LOCK: + if len(files) > _MAX_SKILL_FILES: + raise HostCapabilityError("host_skill_too_large") + if self._storage_root.is_symlink(): + raise HostCapabilityError("invalid_host_skill_path") + self._storage_root.mkdir(parents=True, exist_ok=True, mode=0o700) + target = self._storage_root / descriptor["digest"] + temp = Path(tempfile.mkdtemp(prefix=".skill-", dir=self._storage_root)) + total = 0 + seen: set[PurePosixPath] = set() + try: + for entry in files: + if not isinstance(entry, dict): + raise HostCapabilityError("invalid_host_skill") + raw_path = entry.get("path") + encoding = entry.get("encoding") + raw_content = entry.get("content") + if not isinstance(raw_path, str) or not isinstance( + raw_content, + str, + ): + raise HostCapabilityError("invalid_host_skill") + relative = PurePosixPath(raw_path) + if ( + not raw_path + or "\\" in raw_path + or relative.is_absolute() + or relative in seen + or any(part in {"", ".", ".."} for part in relative.parts) + ): + raise HostCapabilityError("invalid_host_skill_path") + seen.add(relative) + if encoding == "utf-8": + data = raw_content.encode("utf-8") + elif encoding == "base64": + try: + data = base64.b64decode(raw_content, validate=True) + except (ValueError, binascii.Error): + raise HostCapabilityError("invalid_host_skill") from None + else: + raise HostCapabilityError("invalid_host_skill") + total += len(data) + if total > _MAX_SKILL_BYTES: + raise HostCapabilityError("host_skill_too_large") + destination = temp.joinpath(*relative.parts) + destination.parent.mkdir(parents=True, exist_ok=True) + fd = os.open( + destination, + os.O_WRONLY | os.O_CREAT | os.O_EXCL, + 0o600, + ) + with os.fdopen(fd, "wb") as handle: + handle.write(data) + if not (temp / "SKILL.md").is_file(): + raise HostCapabilityError("invalid_host_skill") + if target.is_symlink(): + target.unlink() + elif target.exists(): + shutil.rmtree(target) + os.replace(temp, target) + return target + except HostCapabilityError: + shutil.rmtree(temp, ignore_errors=True) + raise + except OSError: + shutil.rmtree(temp, ignore_errors=True) + raise HostCapabilityError( + "host_skill_materialization_failed", + ) from None + + @staticmethod + def _as_tool_chunk(output: Any) -> ToolChunk: + state = ToolResultState.SUCCESS + blocks: list[str] = [] + chunks = output if isinstance(output, list) else [output] + for chunk in chunks: + if isinstance(chunk, dict) and isinstance(chunk.get("content"), list): + raw_state = chunk.get("state") + if isinstance(raw_state, str) and raw_state in _VALID_STATES: + state = _VALID_STATES[raw_state] + for block in chunk["content"]: + if isinstance(block, dict) and isinstance(block.get("text"), str): + blocks.append(block["text"]) + else: + blocks.append(json.dumps(block, ensure_ascii=False)) + elif isinstance(chunk, str): + blocks.append(chunk) + else: + blocks.append(json.dumps(chunk, ensure_ascii=False)) + return ToolChunk( + content=[TextBlock(text="\n".join(blocks))], + state=state, + ) + + +__all__ = ["HostCapabilityClient", "HostCapabilityError"] diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/task_experience.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/task_experience.py new file mode 100644 index 0000000..bd876fc --- /dev/null +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/task_experience.py @@ -0,0 +1,50 @@ +"""Private Data tools for declaring progress and business deliverables.""" + +from typing import Literal + +from agentscope.message import TextBlock +from agentscope.tool import FunctionTool, ToolChunk + + +def build_task_experience_tools(context_getter): + def runtime(): + # Resolve per call: an Agent/toolkit may survive multiple chat turns. + value = (context_getter() or {}).get("analysis_experience") + if value is None: + raise RuntimeError("analysis experience is unavailable") + return value + + async def report_analysis_progress( + stage: Literal["read_data", "confirm_scope", "analyze", "publish_report"], + ) -> ToolChunk: + """Declare the current analysis stage when work actually changes. + + Use read_data for gathering evidence, confirm_scope for resolving the + analytical definition, analyze for computations/reasoning, and + publish_report only while preparing the chosen user deliverable. + Intermediate file creation alone is not publication or completion. + """ + await runtime().report_analysis_progress(stage) + return ToolChunk(content=[TextBlock(text="Analysis stage recorded.")]) + + async def publish_analysis_artifact( + path: str, + role: Literal["primary", "supporting", "diagnostic", "source"], + kind: Literal["data/report", "data/chart", "data/dataset", "data/diagnostic"], + visibility: Literal["chat", "app_only"] = "chat", + ) -> ToolChunk: + """Publish an existing session artifact as an intentional deliverable. + + Before the final answer, publish the finished report as primary and + only useful charts as supporting. Use app_only for source datasets, + drafts, logs and diagnostics. Files are otherwise retained privately; + file names/extensions never imply publication. path is relative to the + session artifacts directory. This verifies bytes; it cannot create files. + """ + await runtime().publish_analysis_artifact(path, role, kind, visibility) + return ToolChunk(content=[TextBlock(text="Verified artifact published.")]) + + return [ + FunctionTool(report_analysis_progress), + FunctionTool(publish_analysis_artifact), + ] diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/toolkit.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/toolkit.py index 74c921e..ece7be8 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/toolkit.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/toolkit.py @@ -7,6 +7,7 @@ import os import time from collections.abc import Callable +from pathlib import Path from typing import Any from agentscope.tool import Toolkit, ToolGroup @@ -15,6 +16,8 @@ from ..mcp_cm import is_cm_mcp_config, prepare_cm_mcp_clients from ..orchestration import RuntimeStateManager from ..orchestration.tools import PLAN_MODE_TOOL_NAMES, build_qwenpaw_data_tools +from .host_capabilities import HostCapabilityClient +from .task_experience import build_task_experience_tools from .mcp_client_log import MCP_CLIENT_RUN_ID, MCP_RUN_HEADER, log_mcp_client_event from .spawn_subagent import SpawnSubagent from .tools import AskUserQuestionTool, CronJobTool, CronToolServices @@ -283,40 +286,87 @@ async def build_qwenpaw_data_toolkit( cm_mcp_tool_prefixes=cm_mcp_tool_prefixes, ), ) - toolkit = QwenPawDataToolkit( - # AgentScope always enables the basic group, so keep only tools that are - # valid in both Plan Mode and Agent Mode here. - tools=shared_tools, - mcps=_filtered_mcps(workspace_mcps, mcp_catalogs, _is_plan_safe_mcp_tool), - tool_groups=[ - ToolGroup( - name="plan", - description="Tools for creating and revising a QwenPaw Data plan.", - instructions=( - "Use these tools only to create or revise the DAG. " - "You may use MCP metadata/read tools to clarify the plan, " - "but SQL execution tools such as execute_sql are not " - "available in planning mode." - ), + host_capability_client: HostCapabilityClient | None = None + host_skills: list[Any] = [] + request_context = ( + dict(request_context_getter() or {}) + if request_context_getter is not None + else {} + ) + bridge = request_context.get("capability_bridge") + if request_context.get("analysis_experience") is not None: + agent_only_tools.extend(build_task_experience_tools(request_context_getter)) + if bridge is not None: + if not isinstance(bridge, dict): + raise ValueError("invalid_host_capability_bridge") + runtime_dir = workspace_dir or Path.cwd() + storage_dir = getattr(workspace, "host_workdir", None) or runtime_dir + host_capability_client = HostCapabilityClient( + bridge, + storage_workspace_dir=storage_dir, + runtime_workspace_dir=runtime_dir, + ) + try: + shared_tools.extend(await host_capability_client.build_tools()) + host_skills = await host_capability_client.load_skills() + except BaseException: + await host_capability_client.aclose() + raise + try: + toolkit = QwenPawDataToolkit( + # AgentScope always enables the basic group, so keep only tools that are + # valid in both Plan Mode and Agent Mode here. + tools=shared_tools, + skills_or_loaders=host_skills, + mcps=_filtered_mcps( + workspace_mcps, + mcp_catalogs, + _is_plan_safe_mcp_tool, ), - ToolGroup( - name="agent", - description="Tools for executing a QwenPaw Data DAG and producing artifacts.", - instructions=( - "Use update_subtask to record progress for each DAG node. " - "Use workspace tools only for node execution and artifact generation." + tool_groups=[ + ToolGroup( + name="plan", + description=( + "Tools for creating and revising a QwenPaw Data plan." + ), + instructions=( + "Use these tools only to create or revise the DAG. " + "You may use MCP metadata/read tools to clarify the plan, " + "but SQL execution tools such as execute_sql are not " + "available in planning mode." + ), ), - tools=agent_only_tools + workspace_tools, - mcps=_filtered_mcps( - workspace_mcps, - mcp_catalogs, - _is_agent_only_mcp_tool, + ToolGroup( + name="agent", + description=( + "Tools for executing a QwenPaw Data DAG and producing " + "artifacts." + ), + instructions=( + "Use update_subtask to record progress for each DAG node. " + "Use workspace tools only for node execution and artifact " + "generation." + " Report meaningful stage changes with report_analysis_progress." + " Before finishing, use publish_analysis_artifact to select" + " the finished report and useful charts for the user; keep" + " drafts, source data and diagnostics app_only." + ), + tools=agent_only_tools + workspace_tools, + mcps=_filtered_mcps( + workspace_mcps, + mcp_catalogs, + _is_agent_only_mcp_tool, + ), + skills_or_loaders=workspace_skills, ), - skills_or_loaders=workspace_skills, - ), - ], - ) + ], + ) + except BaseException: + if host_capability_client is not None: + await host_capability_client.aclose() + raise toolkit._qwenpaw_data_cm_mcp_tool_prefixes = cm_mcp_tool_prefixes + toolkit._qwenpaw_host_capability_client = host_capability_client toolkit._qwenpaw_context_clients = [ client for client in workspace_mcps diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/app.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/app.py index 173c5e8..176cd1a 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/app.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/app.py @@ -41,6 +41,7 @@ from qwenpaw_data.host.core.api.routers import sessions as sessions_router from qwenpaw_data.host.core.api.routers import settlement as settlement_router from qwenpaw_data.host.core.api.routers import steer as steer_router +from qwenpaw_data.host.core.api.routers import submissions as submissions_router from qwenpaw_data.host.core.api.routers import trace as trace_router from qwenpaw_data.host.core.cron import CronManager from qwenpaw_data.host.core.domain.identity import Identity @@ -105,6 +106,7 @@ async def lifespan(app: FastAPI): reset_runtime_registry() engine = None + submissions_store = None store_mode = (os.environ.get(STORE_ENV) or "").strip().lower() if store_mode == "json": sessions_store: Any = JSONSessionStore(store_root) @@ -131,6 +133,7 @@ async def lifespan(app: FastAPI): SQLSessionStore, SQLSettlementStore, ) + from qwenpaw_data.host.core.store.submissions import SQLSubmissionStore engine, factory = create_engine_and_factory( resolve_db_url(resolved_home), @@ -144,6 +147,9 @@ async def lifespan(app: FastAPI): settlement_store = SQLSettlementStore(factory) attachment_store = SQLAttachmentStore(factory) feedback_store = SQLFeedbackStore(factory) + submissions_store = SQLSubmissionStore(factory) + # Reconcile accepted runs before serving requests or starting jobs. + await submissions_store.recover() async def resolve_model() -> Any: """Prefer the local user's configured default model over env.""" @@ -158,6 +164,17 @@ async def resolve_model() -> Any: ) return build_model_from_env() + async def analysis_model_ready() -> bool: + # Inspect the same local Agent Configuration/env fallback used by + # independent runs. This constructs a model but makes no LLM call. + if model is not None: + return True + try: + await resolve_model() + return True + except Exception: + return False + async def confirmed_settlement_cards(session_id: str) -> list[dict[str, Any]]: session = await sessions_store.get(session_id) return await settlement_store.list_by_session( @@ -184,6 +201,8 @@ def cron_tool_services() -> CronToolServices: settlement=settlement_store, attachments=attachment_store, feedback=feedback_store, + submissions=submissions_store, + analysis_model_ready=analysis_model_ready, hosts=QwenPawDataHostRegistry( home=resolved_home, model=model, @@ -241,6 +260,7 @@ def cron_tool_services() -> CronToolServices: app.add_exception_handler(HTTPException, http_exception_handler) app.include_router(sessions_router.router, prefix="/api/v1") app.include_router(chats_router.router, prefix="/api/v1") + app.include_router(submissions_router.router, prefix="/api/v1") app.include_router(console_router.router, prefix="/api/v1") app.include_router(events_router.router, prefix="/api/v1") app.include_router(plan_router.router, prefix="/api/v1") diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/deps.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/deps.py index 3c9a844..9707ae6 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/deps.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/deps.py @@ -2,8 +2,9 @@ from __future__ import annotations import asyncio -from typing import Any +from typing import Any, TYPE_CHECKING from dataclasses import dataclass, field +from collections.abc import Awaitable, Callable from fastapi import Depends, HTTPException, Request @@ -20,6 +21,9 @@ SettlementStore, ) +if TYPE_CHECKING: + from qwenpaw_data.host.core.store.submissions import SQLSubmissionStore + @dataclass class ServiceState: @@ -32,6 +36,8 @@ class ServiceState: attachments: AttachmentStore feedback: FeedbackStore hosts: QwenPawDataHostRegistry + submissions: SQLSubmissionStore | None = None + analysis_model_ready: Callable[[], Awaitable[bool]] | None = None cron_manager: Any = None tasks: set[asyncio.Task] = field(default_factory=set) diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/artifact.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/artifact.py index 8357db4..45e0178 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/artifact.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/artifact.py @@ -2,16 +2,38 @@ from __future__ import annotations from datetime import datetime +from typing import Literal + +from pydantic import Field from qwenpaw_data.host.core.api.models.common import ApiModel +class ArtifactPresentationSchema(ApiModel): + """An explicit deliverable declaration, independent of its filename.""" + + schema_version: Literal[1] = 1 + role: Literal["primary", "supporting", "diagnostic", "source"] + kind: str = Field( + pattern=r"^[a-z0-9][a-z0-9._-]*/[a-z0-9][a-z0-9._-]*$", max_length=128 + ) + visibility: Literal["chat", "app_only"] = "app_only" + preview: Literal["inline", "link", "none"] = "link" + rank: int = Field(default=100, ge=0, le=10000) + + class ArtifactSchema(ApiModel): id: str session_id: str chat_id: str | None = None name: str path: str + # Added in the artifact-handoff capability. Optional only so historical + # stream rows created by older Engines remain readable. + media_type: str | None = None + size_bytes: int | None = None + digest: str | None = None + presentation: ArtifactPresentationSchema | None = None created_at: datetime updated_at: datetime diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/runtime_models.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/runtime_models.py index 5ad323c..f1b0cc3 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/runtime_models.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/runtime_models.py @@ -1,6 +1,8 @@ -"""Private per-turn bindings accepted by the authenticated native console. +"""Private per-turn bindings for authenticated console and submission APIs. -Credentials are excluded from request serialization and persisted chat records. +Credentials are excluded from request serialization, persisted chat records, +and durable submission digests. Replays retain the original run even when +Host settings have changed. """ from __future__ import annotations diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/stream_objects.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/stream_objects.py index 2529206..b6b3d69 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/stream_objects.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/stream_objects.py @@ -112,6 +112,13 @@ class TaskStatusEventSchema(StreamBaseSchema): graph_snapshot: dict[str, Any] | None = None +class AnalysisProgressEventSchema(StreamBaseSchema): + """Domain-owned progress; normal stream events do not change the stage.""" + + object: Literal["analysis.progress"] = "analysis.progress" + stage: Literal["read_data", "confirm_scope", "analyze", "publish_report"] + + StreamObject = Annotated[ Union[ ResponseEventSchema, @@ -123,6 +130,7 @@ class TaskStatusEventSchema(StreamBaseSchema): ArtifactRegisteredEventSchema, FollowUpGeneratedEventSchema, TaskStatusEventSchema, + AnalysisProgressEventSchema, ], Field(discriminator="object"), ] diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/submissions.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/submissions.py new file mode 100644 index 0000000..48b5eb0 --- /dev/null +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/models/submissions.py @@ -0,0 +1,156 @@ +# -*- coding: utf-8 -*- +from __future__ import annotations + +import hashlib +import json +from typing import Any, Literal +from urllib.parse import urlsplit + +from pydantic import Field, field_validator, model_validator + +from qwenpaw_data.host.core.api.models.chat import AskUserQuestionAnswerSchema +from qwenpaw_data.host.core.api.models.common import ApiModel +from qwenpaw_data.host.core.api.models.artifact import ArtifactCommentSchema +from .runtime_models import RuntimeModels + + +class CapabilityBridgeSchema(ApiModel): + """Short-lived Host callback bound to one App/task identity.""" + + protocol_version: Literal[1] = 1 + endpoint: str = Field(min_length=1, max_length=2048) + token: str = Field(min_length=32, max_length=8192) + + @field_validator("endpoint") + @classmethod + def valid_endpoint(cls, value: str) -> str: + parsed = urlsplit(value) + if ( + parsed.scheme not in {"http", "https"} + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + raise ValueError("endpoint must be an absolute HTTP(S) URL") + return value.rstrip("/") + + +class SubmitRunRequest(ApiModel): + runtime_models: RuntimeModels | None = Field(default=None, exclude=True, repr=False) + protocol_version: Literal[1] = 1 + submission_id: str = Field( + min_length=1, max_length=128, pattern=r"^[A-Za-z0-9_-]+$" + ) + text: str = Field(min_length=1) + datasource_id: str | None = Field(default=None, min_length=1) + dataset_names: list[str] = Field(default_factory=list, max_length=64) + agent_id: str = "default" + capability_bridge: CapabilityBridgeSchema | None = None + session_id: str | None = Field(default=None, min_length=1, max_length=256) + attachment_ids: list[str] = Field(default_factory=list, max_length=32) + artifact_comments: list[ArtifactCommentSchema] = Field( + default_factory=list, max_length=64 + ) + + @field_validator("text", "datasource_id") + @classmethod + def nonblank(cls, value: str | None) -> str | None: + if value is not None and not value.strip(): + raise ValueError("must not be blank") + return value + + @field_validator("dataset_names") + @classmethod + def valid_dataset_names(cls, names: list[str]) -> list[str]: + if any(not name.strip() or len(name) > 256 for name in names): + raise ValueError( + "dataset names must be nonblank and at most 256 characters" + ) + return list(dict.fromkeys(name.strip() for name in names)) + + @model_validator(mode="after") + def require_datasource_or_session(self) -> SubmitRunRequest: + if self.datasource_id is None and self.session_id is None: + raise ValueError("datasource_id is required for a new session") + return self + + def request_digest(self) -> str: + payload = self.model_dump(exclude={"submission_id"}) + # Preserve protocol-1 digests for accepted submissions from older Hosts. + for key in ( + "session_id", + "attachment_ids", + "artifact_comments", + "dataset_names", + ): + if not payload[key]: + payload.pop(key) + encoded = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +class SubmissionRunSchema(ApiModel): + session_id: str + run_id: str + status: Literal[ + "running", + "reconciling", + "succeeded", + "failed", + "cancelled", + "interrupted", + "unknown", + ] + last_sequence_number: int | None = None + error: dict[str, Any] | None = None + reason: str | None = None + + +class SubmissionLookupSchema(ApiModel): + protocol_version: Literal[1] = 1 + submission_id: str + state: Literal["accepted", "not_found"] + run: SubmissionRunSchema | None = None + + +class SubmissionCommandRequest(ApiModel): + protocol_version: Literal[1] = 1 + command_id: str = Field(min_length=1, max_length=128, pattern=r"^[A-Za-z0-9_-]+$") + kind: Literal["answer", "cancel"] + request_id: str | None = Field(default=None, min_length=1, max_length=256) + answers: list[AskUserQuestionAnswerSchema] | None = None + reason: str | None = Field(default=None, max_length=2000) + + @model_validator(mode="after") + def validate_payload(self) -> SubmissionCommandRequest: + if self.kind == "answer": + if not self.request_id or not self.answers or self.reason is not None: + raise ValueError("answer requires request_id and answers") + elif self.request_id is not None or self.answers is not None: + raise ValueError("cancel accepts only an optional reason") + return self + + def request_digest(self) -> str: + encoded = json.dumps( + self.model_dump(exclude={"command_id"}), + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +class SubmissionCommandSchema(ApiModel): + protocol_version: Literal[1] = 1 + submission_id: str + command_id: str + kind: Literal["answer", "cancel"] | None = None + state: Literal["accepted", "rejected", "unknown", "not_found"] + reason: str | None = None diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/artifacts.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/artifacts.py index f4891e4..cd58e0e 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/artifacts.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/artifacts.py @@ -2,34 +2,55 @@ """Session artifact listing and download.""" from __future__ import annotations +import asyncio +import hashlib import os +import re from pathlib import Path from typing import Any from fastapi import APIRouter, Depends, Query -from fastapi.responses import FileResponse +from fastapi.responses import FileResponse, Response -from qwenpaw_data.host.core.api.deps import ServiceState, get_identity, get_state +from qwenpaw_data.host.core.api.deps import ( + ServiceState, + get_identity, + get_state, +) from qwenpaw_data.host.core.api.errors import map_domain_error, raise_api from qwenpaw_data.host.core.domain.identity import Identity from qwenpaw_data.host.core.utils.workspace import list_session_files -router = APIRouter(prefix="/sessions/{session_id}/artifacts", tags=["artifacts"]) +router = APIRouter( + prefix="/sessions/{session_id}/artifacts", + tags=["artifacts"], +) +_DIGEST = re.compile(r"sha256:[0-9a-f]{64}\Z") def _artifact_dir(state: ServiceState, session_id: str) -> Path: return Path(state.hosts.get(session_id=session_id).paths.artifact_dir) +async def _owned_session( + state: ServiceState, + identity: Identity, + session_id: str, +): + session = await state.sessions.get(session_id) + if session.identity.user_id != identity.user_id: + raise LookupError(f"session not found: {session_id}") + return session + + @router.get("") async def list_artifacts( session_id: str, identity: Identity = Depends(get_identity), state: ServiceState = Depends(get_state), ) -> dict[str, Any]: - _ = identity try: - await state.sessions.get(session_id) + await _owned_session(state, identity, session_id) items = list_session_files(_artifact_dir(state, session_id)) except Exception as exc: http = map_domain_error(exc) @@ -43,12 +64,15 @@ async def list_artifacts( async def download_artifact( session_id: str, path: str = Query(..., description="rel_path from the artifact listing"), + digest: str | None = Query( + None, + description="Expected immutable sha256 digest for a bounded copy", + ), identity: Identity = Depends(get_identity), state: ServiceState = Depends(get_state), -) -> FileResponse: - _ = identity +) -> Response: try: - await state.sessions.get(session_id) + await _owned_session(state, identity, session_id) except Exception as exc: http = map_domain_error(exc) if http: @@ -71,6 +95,31 @@ async def download_artifact( normalized = os.path.normpath(os.path.join(str(root), path)) if normalized.startswith(str(root) + os.sep): candidate = Path(normalized).resolve() - if candidate != root and root in candidate.parents and candidate.is_file(): + if ( + candidate != root + and root in candidate.parents + and candidate.is_file() + ): + if digest is not None: + if _DIGEST.fullmatch(digest) is None: + raise_api("NOT_FOUND", "artifact not found", status=404) + try: + content = await asyncio.to_thread(candidate.read_bytes) + except OSError: + raise_api("NOT_FOUND", "artifact not found", status=404) + actual = "sha256:" + hashlib.sha256(content).hexdigest() + if actual != digest: + raise_api( + "CONFLICT", + "artifact version changed", + status=409, + ) + return Response( + content=content, + media_type="application/octet-stream", + headers={ + "X-Artifact-Digest": actual, + }, + ) return FileResponse(candidate, filename=candidate.name) raise_api("NOT_FOUND", "artifact not found", status=404) diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/chats.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/chats.py index f2af3e0..194d93a 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/chats.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/chats.py @@ -69,6 +69,25 @@ async def list_chats( return {"items": [chat_to_schema(c) for c in chats]} +@router.get("/sessions/{session_id}/chats/{chat_id}") +async def get_chat( + session_id: str, + chat_id: str, + identity: Identity = Depends(get_identity), + state: ServiceState = Depends(get_state), +) -> dict[str, Any]: + try: + chat = await state.chats.get(chat_id, session_id=session_id) + if chat.identity.user_id != identity.user_id: + raise LookupError("chat not found") + return {"chat": chat_to_schema(chat)} + except Exception as exc: + http = map_domain_error(exc) + if http: + raise http from exc + raise + + @router.post("/sessions/{session_id}/chats/{chat_id}/stop") async def stop_chat( session_id: str, diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/submissions.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/submissions.py new file mode 100644 index 0000000..7771dc3 --- /dev/null +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/api/routers/submissions.py @@ -0,0 +1,385 @@ +# -*- coding: utf-8 -*- +"""Independent, idempotent runs for the PawApp backend adapter.""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING + +from fastapi import APIRouter, Depends, Header, Query, Request +from fastapi.responses import StreamingResponse + +from qwenpaw_data.host.core.api.deps import ServiceState, get_identity, get_state +from qwenpaw_data.host.core.api.errors import map_domain_error, raise_api +from qwenpaw_data.host.core.api.models.submissions import ( + SubmissionCommandRequest, + SubmissionCommandSchema, + SubmissionLookupSchema, + SubmissionRunSchema, + SubmitRunRequest, +) +from qwenpaw_data.host.core.api.routers.events import chat_events +from qwenpaw_data.host.core.domain.identity import Identity +from qwenpaw_data.host.core.domain.clarification import ( + ClarificationConflict, + ClarificationNotFound, +) +from qwenpaw_data.host.core.runtime.chat_runtime import ChatRuntime +from qwenpaw_data.host.core.runtime.registry import get_runtime_registry +from qwenpaw_data.host.core.stream.output_stream import OutputStream + +if TYPE_CHECKING: + from qwenpaw_data.host.core.store.submissions import ( + SQLSubmissionStore, + SubmissionRecord, + ) + +router = APIRouter(tags=["submissions"]) + + +def _store(state: ServiceState) -> SQLSubmissionStore: + if state.submissions is None: + raise_api( + "VALIDATION", + "Durable submissions require the SQL store", + status=501, + details={"reason": "durable_submissions_unsupported"}, + ) + return state.submissions + + +async def _lookup_response( + store: SQLSubmissionStore, + record: SubmissionRecord, +) -> SubmissionLookupSchema: + return SubmissionLookupSchema( + submission_id=record.submission_id, + state="accepted", + run=SubmissionRunSchema( + session_id=record.session_id, + run_id=record.run_id, + **await store.run_snapshot(record), + ), + ) + + +@router.get("/capabilities/submissions") +async def submission_capabilities( + state: ServiceState = Depends(get_state), +) -> dict: + supported = state.submissions is not None + return { + "protocol_version": 1 if supported else None, + "durable_submissions": supported, + "event_replay": supported, + "durable_commands": supported, + "scoped_host_capabilities": supported, + "artifact_handoff": supported, + "analysis_experience": supported, + "session_submissions": supported, + "optional_session_datasource": supported, + "dataset_context": supported, + "runtime_model_bindings": supported, + "scoped_sessions": True, + } + + +@router.get("/capabilities/analysis") +async def analysis_capabilities( + local_only: bool = False, + identity: Identity = Depends(get_identity), + state: ServiceState = Depends(get_state), +) -> dict: + """Configuration readiness only, never credentials or a provider probe.""" + ready = state.analysis_model_ready + if local_only: + configured = False + if state.prefs is not None: + configured = ( + await state.prefs.load(identity.user_id) + ).runtime_config().default is not None + else: + configured = bool(ready and await ready()) + return { + "readiness_version": 1, + "model_configured": configured, + } + + +@router.get("/sessions/{session_id}/submission-readiness") +async def session_submission_readiness( + session_id: str, + identity: Identity = Depends(get_identity), + state: ServiceState = Depends(get_state), +) -> dict: + try: + session = await state.sessions.get(session_id) + if session.identity.user_id != identity.user_id: + raise LookupError("session not found") + return {"ready": not await state.sessions.has_active_chat(session_id)} + except Exception as exc: + http = map_domain_error(exc) + if http: + raise http from exc + raise + + +async def _accept_and_start( + body: SubmitRunRequest, + identity: Identity, + state: ServiceState, +) -> SubmissionRecord: + # A replay returns the original mapping even after files/session were deleted. + existing = await _store(state).lookup(identity.user_id, body.submission_id) + if existing is not None: + _store(state)._check_digest(existing, body.request_digest()) + return existing + runtime_config = None + context_model_header = None + if body.runtime_models is not None: + from ...domain.preference import UserRuntimeConfig + + local = UserRuntimeConfig() + if state.prefs is not None: + local = (await state.prefs.load(identity.user_id)).runtime_config() + runtime_config = body.runtime_models.runtime_config(local) + context_model_header = body.runtime_models.resolved_context_header( + runtime_config + ) + attachments = [] + if body.session_id is not None: + session = await state.sessions.get(body.session_id) + if session.identity.user_id != identity.user_id: + raise LookupError("session not found") + host = state.hosts.get(session_id=session.id) + attachments = await state.attachments.require_for_session( + identity.user_id, + session.id, + body.attachment_ids, + workspace=host.paths.workspace, + ) + elif body.attachment_ids or body.artifact_comments: + raise ValueError("session context requires an existing session") + record, created = await _store(state).submit( + identity=identity, + submission_id=body.submission_id, + request_digest=body.request_digest(), + text=body.text, + datasource_id=body.datasource_id, + dataset_names=body.dataset_names, + agent_id=body.agent_id, + session_id=body.session_id, + attachments=[item.to_ref() for item in attachments], + artifact_comments=[ + item.model_dump(mode="json") for item in body.artifact_comments + ], + ) + if created: + runtime = ChatRuntime( + chats=state.chats, + events=state.events, + hosts=state.hosts, + prefs=state.prefs, + sessions=state.sessions, + settlement=state.settlement, + ) + run_kwargs = {"identity": identity} + if runtime_config is not None: + run_kwargs["runtime_config"] = runtime_config + run_kwargs["context_model_header"] = context_model_header + if body.capability_bridge is not None: + run_kwargs["capability_bridge"] = body.capability_bridge.model_dump( + mode="json", + ) + state.track( + asyncio.create_task(runtime.run(record.run_id, **run_kwargs)), + ) + return record + + +@router.post("/submissions", response_model=SubmissionLookupSchema, status_code=202) +async def submit_run( + body: SubmitRunRequest, + identity: Identity = Depends(get_identity), + state: ServiceState = Depends(get_state), +) -> SubmissionLookupSchema: + store = _store(state) + # A disconnected HTTP caller must not cancel acceptance between commit and + # scheduling. A process crash in that window is reconciled at startup. + task = asyncio.create_task(_accept_and_start(body, identity, state)) + state.track(task) + try: + record = await asyncio.shield(task) + except Exception as exc: + http = map_domain_error(exc) + if http: + raise http from exc + raise + return await _lookup_response(store, record) + + +@router.get("/submissions/{submission_id}", response_model=SubmissionLookupSchema) +async def lookup_submission( + submission_id: str, + identity: Identity = Depends(get_identity), + state: ServiceState = Depends(get_state), +) -> SubmissionLookupSchema: + store = _store(state) + record = await store.lookup(identity.user_id, submission_id) + if record is None: + return SubmissionLookupSchema(submission_id=submission_id, state="not_found") + return await _lookup_response(store, record) + + +def _command_response(record) -> SubmissionCommandSchema: + return SubmissionCommandSchema( + submission_id=record.submission_id, + command_id=record.command_id, + kind=record.kind, + state=record.state, + reason=record.reason, + ) + + +async def _cancel_submission(record, state: ServiceState) -> str | None: + chat = await state.chats.get(record.run_id, session_id=record.session_id) + runtime = get_runtime_registry().get(record.run_id) + if runtime is not None: + await runtime.cancel() + return None + if chat.status == "running": + # The receipt binds cancellation to this exact run. If its in-memory + # runtime disappeared, persist the same authoritative terminal frame. + await OutputStream( + state.events, + session_id=record.session_id, + chat_id=record.run_id, + identity=chat.identity, + ).response_cancelled() + chat.cancel() + await state.chats.reload_event_watermark(chat) + await state.chats.save(chat) + return None + return "already_terminal" + + +@router.post( + "/submissions/{submission_id}/commands", + response_model=SubmissionCommandSchema, +) +async def execute_submission_command( + submission_id: str, + body: SubmissionCommandRequest, + identity: Identity = Depends(get_identity), + state: ServiceState = Depends(get_state), +) -> SubmissionCommandSchema: + store = _store(state) + submission = await store.lookup(identity.user_id, submission_id) + if submission is None: + raise_api("NOT_FOUND", "Submission not found", status=404) + try: + command, created = await store.prepare_command( + record=submission, + command_id=body.command_id, + kind=body.kind, + request_digest=body.request_digest(), + ) + except Exception as exc: + http = map_domain_error(exc) + if http: + raise http from exc + raise + # Only the durable insert winner may apply the command. Concurrent and + # post-response retries observe the same receipt without another effect. + if not created: + return _command_response(command) + try: + reason = None + if body.kind == "answer": + runtime = get_runtime_registry().get(submission.run_id) + if runtime is None: + raise ClarificationNotFound() + runtime.answer( + clarification_id=body.request_id or "", + result={ + "status": "answered", + "answers": [ + answer.model_dump(exclude_none=True) + for answer in body.answers or [] + ], + }, + ) + else: + reason = await _cancel_submission(submission, state) + command = await store.finish_command( + command, + state="accepted", + reason=reason, + ) + except (ClarificationConflict, ClarificationNotFound): + command = await store.finish_command( + command, + state="rejected", + reason="stale_request", + ) + return _command_response(command) + + +@router.get( + "/submissions/{submission_id}/commands/{command_id}", + response_model=SubmissionCommandSchema, +) +async def lookup_submission_command( + submission_id: str, + command_id: str, + identity: Identity = Depends(get_identity), + state: ServiceState = Depends(get_state), +) -> SubmissionCommandSchema: + store = _store(state) + submission = await store.lookup(identity.user_id, submission_id) + if submission is None: + raise_api("NOT_FOUND", "Submission not found", status=404) + command = await store.lookup_command( + identity.user_id, + submission_id, + command_id, + ) + if command is None: + return SubmissionCommandSchema( + submission_id=submission_id, + command_id=command_id, + state="not_found", + ) + return _command_response(command) + + +@router.get("/submissions/{submission_id}/events") +async def submission_events( + request: Request, + submission_id: str, + identity: Identity = Depends(get_identity), + state: ServiceState = Depends(get_state), + after_sequence_number: int = Query(-1, ge=-1), + last_event_id: str | None = Header(default=None, alias="Last-Event-ID"), +) -> StreamingResponse: + store = _store(state) + record = await store.lookup(identity.user_id, submission_id) + if record is None: + raise_api("NOT_FOUND", "Submission not found", status=404) + snapshot = await store.run_snapshot(record) + if snapshot["status"] == "unknown": + raise_api( + "NOT_FOUND", + "Accepted submission's run is unavailable", + status=410, + details={"reason": "run_missing"}, + ) + return await chat_events( + request, + record.session_id, + record.run_id, + identity, + state, + after_sequence_number, + last_event_id, + ) diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/core.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/core.py index c1518fd..2f74eca 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/core.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/core.py @@ -94,6 +94,7 @@ def __init__( ) self.dag_store = DAGStore(self.paths.dag_root) self._agent: QwenPawDataAgent | None = None + self._host_capability_client: Any = None # ------------------------------------------------------------------ # 配置 / 路径 / 会话访问 @@ -162,6 +163,10 @@ async def close(self) -> None: local workspace 无 close 钩子时为 no-op；重复调用安全。 """ + capability_client = self._host_capability_client + self._host_capability_client = None + if capability_client is not None: + await capability_client.aclose() workspace = self.workspace if workspace is None: return @@ -231,11 +236,17 @@ async def _get_agent( host_artifact_dir=paths.artifact_dir, session_id_getter=lambda: self.session_id, request_context_getter=lambda: dict( - getattr(agent_ref.get("agent"), "_request_context", None) or {}, + getattr(agent_ref.get("agent"), "_request_context", None) + or effective_context, ), enable_clarification=self.enable_clarification, cron_services_factory=self._cron_services_factory, ) + self._host_capability_client = getattr( + toolkit, + "_qwenpaw_host_capability_client", + None, + ) session_store = self.session_store permission_context = build_permission_context( mode=self.permission_mode, diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/db/tables.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/db/tables.py index 3885953..a03385c 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/db/tables.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/db/tables.py @@ -83,6 +83,41 @@ class ChatRow(UserIdColumn, Base): ) +class SubmissionRow(Base): + """Permanent idempotency receipt; never removed with session history.""" + + __tablename__ = "submissions" + + user_id: Mapped[str] = mapped_column(String, primary_key=True) + submission_id: Mapped[str] = mapped_column(String(128), primary_key=True) + request_digest: Mapped[str] = mapped_column(String(64), nullable=False) + session_id: Mapped[str] = mapped_column(String, nullable=False) + run_id: Mapped[str] = mapped_column(String, nullable=False, unique=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=utcnow + ) + + +class SubmissionCommandRow(Base): + """Permanent command identity and outcome for a durable submission.""" + + __tablename__ = "submission_commands" + + user_id: Mapped[str] = mapped_column(String, primary_key=True) + submission_id: Mapped[str] = mapped_column(String(128), primary_key=True) + command_id: Mapped[str] = mapped_column(String(128), primary_key=True) + kind: Mapped[str] = mapped_column(String, nullable=False) + request_digest: Mapped[str] = mapped_column(String(64), nullable=False) + state: Mapped[str] = mapped_column(String, nullable=False, default="prepared") + reason: Mapped[str | None] = mapped_column(String, nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=utcnow + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=utcnow + ) + + class AttachmentRow(UserIdColumn, Base): __tablename__ = "attachments" @@ -211,4 +246,3 @@ class UserActiveModelRow(Base): updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=utcnow ) - diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/runtime/chat_runtime.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/runtime/chat_runtime.py index fae5018..d964d95 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/runtime/chat_runtime.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/runtime/chat_runtime.py @@ -2,7 +2,9 @@ from __future__ import annotations import asyncio +import hashlib import logging +import mimetypes import os from datetime import datetime, timezone from functools import partial @@ -15,8 +17,10 @@ from qwenpaw_data.host.core.algo.biztrace.transformer import BizTraceTransformer from qwenpaw_data.host.core.algo.followup.recommend import FollowUpRecommend from qwenpaw_data.host.core.algo.settlement import SettlementManager, SettlementSettings -from qwenpaw_data.host.core.domain.chat import Chat from qwenpaw_data.host.core.domain.preference import UserRuntimeConfig +from qwenpaw_data.host.core.domain.chat import Chat +from qwenpaw_data.host.core.api.models.artifact import ArtifactPresentationSchema +from qwenpaw_data.host.core.artifact_paths import ArtifactPathContext from qwenpaw_data.host.core.domain.identity import Identity from qwenpaw_data.host.core.orchestration.tools import PLAN_TOOL_NAMES from qwenpaw_data.host.core.registry import QwenPawDataHostRegistry @@ -113,6 +117,7 @@ async def run( chat_id: str, *, identity: Identity, + capability_bridge: dict[str, Any] | None = None, runtime_config: UserRuntimeConfig | None = None, context_model_header: str | None = None, ) -> None: @@ -122,6 +127,7 @@ async def run( await self._run( chat_id, identity=identity, + capability_bridge=capability_bridge, runtime_config=runtime_config, context_model_header=context_model_header, ) @@ -143,6 +149,7 @@ async def _run( chat_id: str, *, identity: Identity, + capability_bridge: dict[str, Any] | None = None, runtime_config: UserRuntimeConfig | None = None, context_model_header: str | None = None, ) -> None: @@ -169,10 +176,14 @@ async def _run( request_context = { "datasource_id": chat.datasource_id, "user_id": chat.identity.user_id, + "analysis_experience": self, } + if capability_bridge is not None: + request_context["capability_bridge"] = dict(capability_bridge) if runtime_config is not None: from ..providers.factory import build_model - + # Sessions serialize turns; changing the model here preserves memory + # and never modifies a currently running turn. host.model = build_model(runtime_config.default) if host._agent is not None: host._agent.model = host.model @@ -198,6 +209,8 @@ async def _run( identity=identity, request_context=request_context, ) + self._artifact_seen = self._artifact_snapshot() + await self.report_analysis_progress("read_data") if runtime_config is not None: self._run_context.user_runtime_config = runtime_config else: @@ -383,7 +396,7 @@ async def _load_previous_followups(self, chat: Chat) -> tuple[str, ...]: return () async def _after_event_callback(self, event: Any) -> None: - if self._biztrace is not None and self._is_tool_success(event): + if self._is_tool_success(event): # Delta first: the pipeline binds pending files to the coming # TOOL_RESULT_END, so order matters. await self._register_new_files() @@ -413,9 +426,8 @@ def _is_tool_success(event: Any) -> bool: async def _register_new_files(self) -> None: """Diff the artifact directory; notify the stream and the algorithm.""" - biztrace = self._biztrace envelope = self._envelope - if biztrace is None or envelope is None: + if envelope is None: return try: current = self._artifact_snapshot() @@ -431,23 +443,126 @@ async def _register_new_files(self) -> None: files: dict[str, str] = {} for path in changed: name = Path(path).name + metadata = await asyncio.to_thread( + self._artifact_metadata, + path, + ) + if metadata is None: + continue await envelope.stream.artifact_registered( id=create_id("artifact"), name=name, path=path, + **metadata, + presentation={ + "role": "diagnostic", + "kind": "data/diagnostic", + "visibility": "app_only", + "preview": "none", + "rank": 300, + }, created_at=now, updated_at=now, ) files[name] = path - await biztrace.append( - {"kind": "artifact_delta", "payload": {"files": files}} - ) + if files and self._biztrace is not None: + await self._biztrace.append( + {"kind": "artifact_delta", "payload": {"files": files}} + ) except Exception: logger.exception( - "biztrace: failed to register artifact files for chat %s", - biztrace.chat_id, + "failed to register artifact files for chat %s", + self._run_context.chat_id if self._run_context else "unknown", ) + async def report_analysis_progress(self, stage: str) -> None: + if self._envelope is None: + raise RuntimeError("analysis stream is unavailable") + # OutputStream validates and persists the typed event before returning. + await self._envelope.stream.append( + {"object": "analysis.progress", "stage": stage} + ) + + async def publish_analysis_artifact( + self, + path: str, + role: str, + kind: str, + visibility: str, + ) -> None: + ctx, envelope = self._run_context, self._envelope + if ctx is None or envelope is None: + raise RuntimeError("analysis stream is unavailable") + root = Path(ctx.paths.artifact_dir) + resolved = ArtifactPathContext(root, root).resolve_ref(path) + if not resolved.host_path.is_file(): + raise ValueError("artifact must be an existing session file") + metadata = await asyncio.to_thread( + self._artifact_metadata, resolved.relative_path + ) + if metadata is None: + raise ValueError("artifact changed during verification; retry publication") + presentation = ArtifactPresentationSchema( + role=role, + kind=kind, + visibility=visibility, + preview=( + "inline" + if metadata["media_type"] + in { + "text/html", + "text/markdown", + "text/plain", + } + and visibility == "chat" + else "link" if visibility == "chat" else "none" + ), + rank={"primary": 0, "supporting": 20, "source": 200, "diagnostic": 300}[ + role + ], + ) + now = datetime.now(timezone.utc) + await envelope.stream.artifact_registered( + id=create_id("artifact"), + name=resolved.host_path.name, + path=resolved.relative_path, + **metadata, + presentation=presentation.model_dump(mode="json"), + created_at=now, + updated_at=now, + ) + # Do not immediately republish the same bytes as an automatic diagnostic. + stamp = self._artifact_snapshot().get(resolved.relative_path) + if stamp is not None: + self._artifact_seen[resolved.relative_path] = stamp + + def _artifact_metadata(self, path: str) -> dict[str, Any] | None: + ctx = self._run_context + if ctx is None: + return None + root = Path(ctx.paths.artifact_dir) + candidate = ArtifactPathContext(root, root).resolve_path(path) + try: + before = candidate.stat() + digest = hashlib.sha256() + with candidate.open("rb") as stream: + while chunk := stream.read(1024 * 1024): + digest.update(chunk) + after = candidate.stat() + except OSError: + return None + if (before.st_size, before.st_mtime_ns) != ( + after.st_size, + after.st_mtime_ns, + ): + return None + media_type = mimetypes.guess_type(candidate.name)[0] + return { + "media_type": media_type or "application/octet-stream", + "size_bytes": after.st_size, + "digest": f"sha256:{digest.hexdigest()}", + } + def _artifact_snapshot(self) -> dict[str, tuple[int, int]]: ctx = self._run_context if ctx is None: diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/store/sql_store.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/store/sql_store.py index 6bce649..565b3bc 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/store/sql_store.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/store/sql_store.py @@ -318,6 +318,23 @@ def _apply_session(row: SessionRow, session: Session) -> None: row.deleted_at = session.deleted_at +def _session_to_row(session: Session) -> SessionRow: + return SessionRow( + id=session.id, + user_id=session.identity.user_id, + agent_id=session.agent_id, + title=session.title, + datasource_id=session.datasource_id, + channel=session.channel, + chat_count=session.chat_count, + parent_session_id=session.parent_session_id, + forked_from_chat_id=session.forked_from_chat_id, + created_at=session.created_at, + updated_at=session.updated_at, + deleted_at=session.deleted_at, + ) + + class SQLSessionStore: def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: self._sessions = session_factory @@ -326,25 +343,8 @@ async def add(self, session: Session) -> None: async with self._sessions() as db: existing = await db.get(SessionRow, session.id) if existing is not None: - raise RuntimeError( - f"CONFLICT: session already exists: {session.id}" - ) - db.add( - SessionRow( - id=session.id, - user_id=session.identity.user_id, - agent_id=session.agent_id, - title=session.title, - datasource_id=session.datasource_id, - channel=session.channel, - chat_count=session.chat_count, - parent_session_id=session.parent_session_id, - forked_from_chat_id=session.forked_from_chat_id, - created_at=session.created_at, - updated_at=session.updated_at, - deleted_at=session.deleted_at, - ) - ) + raise RuntimeError(f"CONFLICT: session already exists: {session.id}") + db.add(_session_to_row(session)) await db.commit() async def get(self, session_id: str) -> Session: @@ -479,6 +479,30 @@ def _apply_chat(row: ChatRow, chat: Chat) -> None: row.updated_at = chat.updated_at +def _chat_to_row(chat: Chat) -> ChatRow: + return ChatRow( + id=chat.id, + session_id=chat.session_id, + user_id=chat.identity.user_id, + sequence=chat.sequence, + user_input=chat.user_input, + datasource_id=chat.datasource_id, + kind=chat.kind, + status=chat.status, + last_sequence_number=chat.last_sequence_number, + started_at=chat.started_at, + completed_at=chat.completed_at, + active_duration_ms=chat.active_duration_ms, + error_json=chat.error, + plan=chat.plan, + artifact_comments_json=list(chat.artifact_comments), + attachments_json=list(chat.attachments), + dataset_names_json=list(chat.dataset_names), + created_at=chat.created_at, + updated_at=chat.updated_at, + ) + + class SQLChatStore: def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: self._sessions = session_factory @@ -488,29 +512,7 @@ async def add(self, chat: Chat) -> None: existing = await db.get(ChatRow, chat.id) if existing is not None: raise RuntimeError(f"CONFLICT: chat already exists: {chat.id}") - db.add( - ChatRow( - id=chat.id, - session_id=chat.session_id, - user_id=chat.identity.user_id, - sequence=chat.sequence, - user_input=chat.user_input, - datasource_id=chat.datasource_id, - kind=chat.kind, - status=chat.status, - last_sequence_number=chat.last_sequence_number, - started_at=chat.started_at, - completed_at=chat.completed_at, - active_duration_ms=chat.active_duration_ms, - error_json=chat.error, - plan=chat.plan, - artifact_comments_json=list(chat.artifact_comments), - attachments_json=list(chat.attachments), - dataset_names_json=list(chat.dataset_names), - created_at=chat.created_at, - updated_at=chat.updated_at, - ) - ) + db.add(_chat_to_row(chat)) await db.commit() async def get( diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/store/submissions.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/store/submissions.py new file mode 100644 index 0000000..7493f7c --- /dev/null +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/store/submissions.py @@ -0,0 +1,370 @@ +# -*- coding: utf-8 -*- +"""Durable submission receipts for trusted backend callers. + +Acceptance is atomic with session/run creation. Only the insert winner may +schedule execution; a receipt is never discarded or reused after a restart. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from sqlalchemy import exists, or_, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from qwenpaw_data.host.core.api.models.stream_objects import ( + dump_stream_object, + parse_stream_object, +) +from qwenpaw_data.host.core.db.tables import ( + ChatEventRow, + ChatRow, + SubmissionCommandRow, + SubmissionRow, + SessionRow, +) +from qwenpaw_data.host.core.domain.identity import Identity +from qwenpaw_data.host.core.domain.session import Session +from qwenpaw_data.host.core.store.sql_store import ( + _chat_to_row, + _session_to_row, + _session_from_row, + _apply_session, +) +from qwenpaw_data.host.core.utils.ids import create_id +from qwenpaw_data.host.core.utils.time import utcnow + +_TERMINAL = ("completed", "failed", "cancelled") +_RESTART_REASON = "executor_restarted" + + +@dataclass(frozen=True) +class SubmissionRecord: + user_id: str + submission_id: str + request_digest: str + session_id: str + run_id: str + + +@dataclass(frozen=True) +class SubmissionCommandRecord: + user_id: str + submission_id: str + command_id: str + kind: str + request_digest: str + state: str + reason: str | None + + +def _record(row: SubmissionRow) -> SubmissionRecord: + return SubmissionRecord( + row.user_id, row.submission_id, row.request_digest, row.session_id, row.run_id + ) + + +def _command_record(row: SubmissionCommandRow) -> SubmissionCommandRecord: + return SubmissionCommandRecord( + row.user_id, + row.submission_id, + row.command_id, + row.kind, + row.request_digest, + "unknown" if row.state == "prepared" else row.state, + row.reason, + ) + + +def _terminal_events(chat_id): + return select(ChatEventRow).where( + ChatEventRow.chat_id == chat_id, + ChatEventRow.object == "response", + ChatEventRow.payload["status"].as_string().in_(_TERMINAL), + ) + + +class SQLSubmissionStore: + def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: + self._sessions = session_factory + + async def lookup(self, user_id: str, submission_id: str) -> SubmissionRecord | None: + async with self._sessions() as db: + row = await db.get(SubmissionRow, (user_id, submission_id)) + return _record(row) if row is not None else None + + async def lookup_command( + self, + user_id: str, + submission_id: str, + command_id: str, + ) -> SubmissionCommandRecord | None: + async with self._sessions() as db: + row = await db.get( + SubmissionCommandRow, + (user_id, submission_id, command_id), + ) + return _command_record(row) if row is not None else None + + async def prepare_command( + self, + *, + record: SubmissionRecord, + command_id: str, + kind: str, + request_digest: str, + ) -> tuple[SubmissionCommandRecord, bool]: + existing = await self.lookup_command( + record.user_id, + record.submission_id, + command_id, + ) + if existing is not None: + if existing.kind != kind or existing.request_digest != request_digest: + raise RuntimeError("CONFLICT: command_id was used for different inputs") + return existing, False + row = SubmissionCommandRow( + user_id=record.user_id, + submission_id=record.submission_id, + command_id=command_id, + kind=kind, + request_digest=request_digest, + state="prepared", + ) + try: + async with self._sessions() as db, db.begin(): + db.add(row) + await db.flush() + except IntegrityError: + existing = await self.lookup_command( + record.user_id, + record.submission_id, + command_id, + ) + if existing is None: + raise + if existing.kind != kind or existing.request_digest != request_digest: + raise RuntimeError( + "CONFLICT: command_id was used for different inputs" + ) from None + return existing, False + return _command_record(row), True + + async def finish_command( + self, + record: SubmissionCommandRecord, + *, + state: str, + reason: str | None = None, + ) -> SubmissionCommandRecord: + if state not in {"accepted", "rejected"}: + raise ValueError("invalid command outcome") + async with self._sessions() as db, db.begin(): + row = await db.get( + SubmissionCommandRow, + (record.user_id, record.submission_id, record.command_id), + ) + if row is None: + raise LookupError("submission command not found") + if row.state != "prepared": + return _command_record(row) + row.state = state + row.reason = reason + row.updated_at = utcnow() + await db.flush() + return _command_record(row) + + @staticmethod + def _check_digest(record: SubmissionRecord, digest: str) -> None: + if record.request_digest != digest: + raise RuntimeError("CONFLICT: submission_id was used for different inputs") + + async def submit( + self, + *, + identity: Identity, + submission_id: str, + request_digest: str, + text: str, + datasource_id: str | None, + agent_id: str, + session_id: str | None = None, + dataset_names: list[str] | None = None, + attachments: list[dict] | None = None, + artifact_comments: list[dict] | None = None, + ) -> tuple[SubmissionRecord, bool]: + existing = await self.lookup(identity.user_id, submission_id) + if existing is not None: + self._check_digest(existing, request_digest) + return existing, False + + row = SubmissionRow( + user_id=identity.user_id, + submission_id=submission_id, + request_digest=request_digest, + session_id=session_id or create_id("ses"), + run_id=create_id("chat"), + ) + try: + async with self._sessions() as db, db.begin(): + db.add(row) + # The DB unique key arbitrates concurrent submissions. Do not + # call runtime/model code until all three rows have committed. + await db.flush() + session_row = None + if session_id is not None: + session_row = await db.scalar( + select(SessionRow) + .where( + SessionRow.id == session_id, + SessionRow.user_id == identity.user_id, + SessionRow.deleted_at.is_(None), + ) + .with_for_update() + ) + if session_row is None: + raise LookupError("session not found") + session = _session_from_row(session_row) + else: + session = Session.create( + identity=identity, + agent_id=agent_id, + datasource_id=datasource_id, + channel="pawapp", + ) + session.id = row.session_id + active = await db.scalar( + select( + exists().where( + ChatRow.session_id == session.id, + ChatRow.status == "running", + ) + ) + ) + chat = session.open_chat( + text=text, + datasource_id=datasource_id, + has_active_chat=bool(active), + dataset_names=dataset_names, + attachments=attachments, + artifact_comments=artifact_comments, + ) + chat.id = row.run_id + if session_row is not None: + _apply_session(session_row, session) + else: + db.add(_session_to_row(session)) + db.add(_chat_to_row(chat)) + except IntegrityError: + existing = await self.lookup(identity.user_id, submission_id) + if existing is None: + raise # Some other integrity violation; not a deduplication hit. + self._check_digest(existing, request_digest) + return existing, False + return _record(row), True + + async def run_snapshot(self, record: SubmissionRecord) -> dict[str, Any]: + async with self._sessions() as db: + chat = await db.get(ChatRow, record.run_id) + if ( + chat is None + or chat.session_id != record.session_id + or chat.user_id != record.user_id + ): + return {"status": "unknown", "reason": "run_missing"} + terminal = await db.scalar( + _terminal_events(chat.id) + .order_by(ChatEventRow.sequence_number.desc()) + .limit(1) + ) + status = "running" if chat.status == "running" else "reconciling" + error = chat.error_json + if terminal is not None: + status = { + "completed": "succeeded", + "failed": "failed", + "cancelled": "cancelled", + }[terminal.payload["status"]] + error = terminal.payload.get("error") + if ((error or {}).get("details") or {}).get( + "reason" + ) == _RESTART_REASON: + status = "interrupted" + return { + "status": status, + "last_sequence_number": max( + chat.last_sequence_number, + terminal.sequence_number if terminal is not None else -1, + ), + "error": error, + } + + async def recover(self) -> None: + """Startup only, before any writers: persist an explicit terminal fact. + + A terminal response is the success boundary. A saved completed chat + without that response may have lost in-memory output finalization, so + it is interrupted, never synthesized as a successful execution. + """ + async with self._sessions() as db, db.begin(): + chats = ( + await db.scalars( + select(ChatRow) + .join(SubmissionRow, SubmissionRow.run_id == ChatRow.id) + .where( + or_( + ChatRow.status == "running", + ~exists(_terminal_events(ChatRow.id)), + ) + ) + ) + ).all() + for chat in chats: + terminal = await db.scalar( + _terminal_events(chat.id) + .order_by(ChatEventRow.sequence_number.desc()) + .limit(1) + ) + now = utcnow() + if terminal is not None: + chat.status = { + "completed": "completed", + "failed": "failed", + "cancelled": "canceled", + }[terminal.payload["status"]] + chat.error_json = terminal.payload.get("error") + else: + if chat.status not in ("failed", "canceled"): + chat.error_json = { + "code": "VALIDATION", + "message": "Engine restarted before a terminal response was persisted", + "details": { + "reason": _RESTART_REASON, + "previous_status": chat.status, + }, + } + chat.status = "canceled" + chat.last_sequence_number += 1 + payload = { + "object": "response", + "id": create_id("response"), + "status": "failed" if chat.status == "failed" else "cancelled", + "session_id": chat.session_id, + "chat_id": chat.id, + "sequence_number": chat.last_sequence_number, + "error": chat.error_json, + } + db.add( + ChatEventRow( + chat_id=chat.id, + session_id=chat.session_id, + user_id=chat.user_id, + sequence_number=chat.last_sequence_number, + object="response", + payload=dump_stream_object(parse_stream_object(payload)), + ) + ) + chat.completed_at = chat.completed_at or now + chat.updated_at = now diff --git a/packages/qwenpaw-data-host-core/tests/test_artifact_registration.py b/packages/qwenpaw-data-host-core/tests/test_artifact_registration.py new file mode 100644 index 0000000..3796e37 --- /dev/null +++ b/packages/qwenpaw-data-host-core/tests/test_artifact_registration.py @@ -0,0 +1,83 @@ +# -*- coding: utf-8 -*- +"""Artifact stream registration remains available without BizTrace.""" + +from __future__ import annotations + +import hashlib +from types import SimpleNamespace +import pytest + +from qwenpaw_data.host.core.domain.identity import Identity +from qwenpaw_data.host.core.runtime.chat_runtime import ChatRuntime +from qwenpaw_data.host.core.runtime.context import RunContext + + +class _Stream: + def __init__(self): + self.items = [] + + async def artifact_registered(self, **fields): + self.items.append(fields) + + async def append(self, fields): + self.items.append(fields) + + +async def test_registers_digest_metadata_when_biztrace_is_disabled(tmp_path): + runtime = ChatRuntime(chats=object(), events=object(), hosts=object()) + stream = _Stream() + runtime._envelope = SimpleNamespace(stream=stream) + runtime._run_context = RunContext( + session_id="session-1", + chat_id="chat-1", + workspace=tmp_path, + paths=SimpleNamespace(artifact_dir=tmp_path), + identity=Identity(user_id="alice"), + ) + content = b"# report\n" + (tmp_path / "report.md").write_bytes(content) + + await runtime._register_new_files() + + assert len(stream.items) == 1 + artifact = stream.items[0] + assert artifact["path"] == "report.md" + assert artifact["media_type"] == "text/markdown" + assert artifact["size_bytes"] == len(content) + assert artifact["digest"] == ("sha256:" + hashlib.sha256(content).hexdigest()) + assert artifact["presentation"]["visibility"] == "app_only" + + +async def test_explicit_deliverable_does_not_depend_on_filename(tmp_path): + runtime = ChatRuntime(chats=object(), events=object(), hosts=object()) + stream = _Stream() + runtime._envelope = SimpleNamespace(stream=stream) + runtime._run_context = RunContext( + session_id="s", + chat_id="c", + workspace=tmp_path, + paths=SimpleNamespace(artifact_dir=tmp_path), + identity=Identity(user_id="alice"), + ) + (tmp_path / "march.html").write_text("

Findings

") + await runtime.publish_analysis_artifact( + "march.html", "primary", "data/report", "chat" + ) + await runtime._register_new_files() + assert len(stream.items) == 1 + assert stream.items[0]["presentation"]["role"] == "primary" + assert stream.items[0]["presentation"]["visibility"] == "chat" + await runtime.report_analysis_progress("publish_report") + assert stream.items[-1] == { + "object": "analysis.progress", + "stage": "publish_report", + } + with pytest.raises(ValueError): + await runtime.publish_analysis_artifact( + "../outside.txt", "primary", "data/report", "chat" + ) + (tmp_path / "escape").symlink_to(tmp_path.parent) + with pytest.raises(ValueError): + await runtime.publish_analysis_artifact( + "escape/outside.txt", "primary", "data/report", "chat" + ) diff --git a/packages/qwenpaw-data-host-core/tests/test_artifacts_api.py b/packages/qwenpaw-data-host-core/tests/test_artifacts_api.py index a808870..49249d9 100644 --- a/packages/qwenpaw-data-host-core/tests/test_artifacts_api.py +++ b/packages/qwenpaw-data-host-core/tests/test_artifacts_api.py @@ -5,6 +5,7 @@ from contextlib import asynccontextmanager from pathlib import Path +import hashlib import pytest @@ -65,6 +66,28 @@ async def test_list_and_download(tmp_path, monkeypatch) -> None: assert nested.status_code == 200 assert nested.text == "a,b\n1,2\n" + digest = "sha256:" + hashlib.sha256(b"png-bytes").hexdigest() + immutable = await http.get( + f"/api/v1/sessions/{session_id}/artifacts/file", + params={"path": "chart.png", "digest": digest}, + ) + assert immutable.status_code == 200 + assert immutable.content == b"png-bytes" + assert immutable.headers["x-artifact-digest"] == digest + + changed = await http.get( + f"/api/v1/sessions/{session_id}/artifacts/file", + params={"path": "chart.png", "digest": "sha256:" + "0" * 64}, + ) + assert changed.status_code == 409 + + forbidden = await http.get( + f"/api/v1/sessions/{session_id}/artifacts/file", + params={"path": "chart.png"}, + headers={"X-User-Id": "another-user"}, + ) + assert forbidden.status_code == 404 + async def test_download_rejects_traversal_and_missing( tmp_path, monkeypatch diff --git a/packages/qwenpaw-data-host-core/tests/test_host_capabilities.py b/packages/qwenpaw-data-host-core/tests/test_host_capabilities.py new file mode 100644 index 0000000..a8bdcea --- /dev/null +++ b/packages/qwenpaw-data-host-core/tests/test_host_capabilities.py @@ -0,0 +1,231 @@ +from __future__ import annotations + +import json + +import httpx +import pytest +from agentscope.message import ToolResultState + +from qwenpaw_data.host.core.agent.host_capabilities import ( + HostCapabilityClient, + HostCapabilityError, +) + + +def descriptor( + capability_id: str, + *, + kind: str, + wire_name: str, + digest: str, + available: bool = True, + blocked_reason: str | None = None, +) -> dict: + return { + "capability_id": capability_id, + "name": capability_id.rsplit("/", 1)[-1], + "wire_name": wire_name, + "kind": kind, + "scope": "app_private", + "description": f"{wire_name} description", + "input_schema": ( + { + "type": "object", + "properties": {"text": {"type": "string"}}, + "required": ["text"], + } + if kind == "tool" + else None + ), + "tool_refs": [], + "is_read_only": kind == "tool", + "digest": digest, + "available": available, + "blocked_reason": blocked_reason, + } + + +def client(tmp_path, handler) -> HostCapabilityClient: + return HostCapabilityClient( + { + "protocol_version": 1, + "endpoint": "http://host.test/api/pawapp-capabilities", + "token": "scoped-token", + }, + storage_workspace_dir=tmp_path, + runtime_workspace_dir="/workspace", + transport=httpx.MockTransport(handler), + ) + + +async def test_remote_tool_and_skill_are_loaded_with_scoped_auth(tmp_path) -> None: + calls: list[tuple[str, str]] = [] + tool = descriptor( + "app/tool/echo", + kind="tool", + wire_name="app__echo", + digest="a" * 64, + ) + skill = descriptor( + "app/skill/guidance", + kind="skill", + wire_name="guidance", + digest="b" * 64, + ) + + def handler(request: httpx.Request) -> httpx.Response: + assert request.headers["authorization"] == "Bearer scoped-token" + calls.append((request.method, request.url.raw_path.decode())) + if request.url.path.endswith("/catalog"): + return httpx.Response( + 200, + json={"protocol_version": 1, "capabilities": [tool, skill]}, + ) + if "/tools/" in request.url.path: + assert json.loads(request.content) == {"params": {"text": "hello"}} + return httpx.Response( + 200, + json={ + "protocol_version": 1, + "state": "success", + "output": { + "content": [{"type": "text", "text": "echo: hello"}], + "state": "success", + }, + }, + ) + return httpx.Response( + 200, + json={ + "protocol_version": 1, + "descriptor": skill, + "files": [ + { + "path": "SKILL.md", + "encoding": "utf-8", + "content": ( + "---\nname: guidance\n" + "description: Use the echo tool\n---\n" + "Call app__echo.\n" + ), + }, + { + "path": "references/example.txt", + "encoding": "utf-8", + "content": "example", + }, + ], + }, + ) + + bridge = client(tmp_path, handler) + try: + tools = await bridge.build_tools() + skills = await bridge.load_skills() + result = await tools[0](text="hello") + finally: + await bridge.aclose() + + assert tools[0].name == "app__echo" + assert tools[0].is_read_only is True + assert result.state is ToolResultState.SUCCESS + assert result.content[0].text == "echo: hello" + assert skills[0].name == "guidance" + assert skills[0].dir == f"/workspace/.pawapp-capabilities/{'b' * 64}" + materialized = tmp_path / ".pawapp-capabilities" / ("b" * 64) + assert (materialized / "references/example.txt").read_text() == "example" + assert sum(path.endswith("/catalog") for _, path in calls) == 1 + assert any("app%2Ftool%2Fecho" in path for _, path in calls) + + +async def test_remote_denial_becomes_denied_tool_result(tmp_path) -> None: + tool = descriptor( + "host/tool/bash", + kind="tool", + wire_name="bash", + digest="c" * 64, + ) + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/catalog"): + return httpx.Response( + 200, + json={"protocol_version": 1, "capabilities": [tool]}, + ) + return httpx.Response(403, json={"detail": "tool_permission_denied"}) + + bridge = client(tmp_path, handler) + try: + remote = (await bridge.build_tools())[0] + result = await remote(text="unsafe") + finally: + await bridge.aclose() + + assert result.state is ToolResultState.DENIED + assert result.content[0].text == "HostCapabilityError: tool_permission_denied" + + +async def test_skill_materialization_rejects_path_traversal(tmp_path) -> None: + skill = descriptor( + "host/skill/guidance", + kind="skill", + wire_name="guidance", + digest="d" * 64, + ) + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/catalog"): + return httpx.Response( + 200, + json={"protocol_version": 1, "capabilities": [skill]}, + ) + return httpx.Response( + 200, + json={ + "protocol_version": 1, + "descriptor": skill, + "files": [ + { + "path": "../outside.txt", + "encoding": "utf-8", + "content": "escape", + }, + ], + }, + ) + + bridge = client(tmp_path, handler) + try: + with pytest.raises(HostCapabilityError, match="invalid_host_skill_path"): + await bridge.load_skills() + finally: + await bridge.aclose() + assert not (tmp_path / "outside.txt").exists() + + +async def test_unavailable_skill_dependency_fails_before_loading(tmp_path) -> None: + skill = descriptor( + "host/skill/guidance", + kind="skill", + wire_name="guidance", + digest="e" * 64, + available=False, + blocked_reason="skill_dependency_unavailable", + ) + + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path.endswith("/catalog") + return httpx.Response( + 200, + json={"protocol_version": 1, "capabilities": [skill]}, + ) + + bridge = client(tmp_path, handler) + try: + with pytest.raises( + HostCapabilityError, + match="skill_dependency_unavailable", + ): + await bridge.load_skills() + finally: + await bridge.aclose() diff --git a/packages/qwenpaw-data-host-core/tests/test_host_runtime.py b/packages/qwenpaw-data-host-core/tests/test_host_runtime.py index ba7efb0..277159f 100644 --- a/packages/qwenpaw-data-host-core/tests/test_host_runtime.py +++ b/packages/qwenpaw-data-host-core/tests/test_host_runtime.py @@ -120,11 +120,42 @@ async def fake_build_toolkit(*args: Any, **kwargs: Any) -> object: assert toolkit_kwargs["workspace_dir"] == PurePosixPath("/workspace") assert toolkit_kwargs["artifacts_root"] == PurePosixPath("/workspace/artifacts") assert toolkit_kwargs["host_artifact_dir"] == host.paths.artifact_dir + assert toolkit_kwargs["request_context_getter"]() == {} artifact_context = agent.kwargs["runtime_state"]._artifact_path_context assert artifact_context.host_artifact_dir == host.paths.artifact_dir.resolve() assert artifact_context.model_artifact_dir == "/workspace/artifacts/session-a" +@pytest.mark.asyncio +async def test_initial_request_context_reaches_toolkit( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + toolkit_kwargs: dict[str, Any] = {} + + async def fake_build_toolkit(*args: Any, **kwargs: Any) -> object: + toolkit_kwargs.update(kwargs) + assert kwargs["request_context_getter"]() == { + "capability_bridge": {"protocol_version": 1}, + } + return object() + + monkeypatch.setattr(core_module, "build_qwenpaw_data_toolkit", fake_build_toolkit) + monkeypatch.setattr(core_module, "QwenPawDataAgent", FakeAgent) + host = QwenPawDataHost( + home=tmp_path, + model=object(), + workspace=FakeWorkspace(str(tmp_path / "workspace")), + ) + + await host._get_agent( + mode="agent", + request_context={"capability_bridge": {"protocol_version": 1}}, + ) + + assert "request_context_getter" in toolkit_kwargs + + @pytest.mark.asyncio async def test_model_paths_fall_back_without_workspace_workdir( tmp_path: Path, diff --git a/packages/qwenpaw-data-host-core/tests/test_preferences_api.py b/packages/qwenpaw-data-host-core/tests/test_preferences_api.py index 83db327..7138850 100644 --- a/packages/qwenpaw-data-host-core/tests/test_preferences_api.py +++ b/packages/qwenpaw-data-host-core/tests/test_preferences_api.py @@ -64,6 +64,53 @@ async def test_provider_config_roundtrip(tmp_path, monkeypatch) -> None: ).status_code == 404 +async def test_analysis_readiness_uses_local_agent_settings_not_caller_prefs( + tmp_path, monkeypatch, +) -> None: + built = [] + + def no_env_model(): + raise RuntimeError("sensitive configuration detail") + + monkeypatch.setattr( + "qwenpaw_data.host.core.api.app.build_model_from_env", no_env_model, + ) + monkeypatch.setattr( + "qwenpaw_data.host.core.api.app.build_model", + lambda active: built.append(active) or object(), + ) + async with service_client(tmp_path, monkeypatch) as (http, _): + url = "/api/v1/capabilities/analysis" + missing = await http.get(url, headers={"X-User-Id": "pawapp-scoped"}) + assert missing.json() == {"readiness_version": 1, "model_configured": False} + assert "sensitive" not in missing.text + await http.put( + "/api/v1/preferences/providers/dashscope", + json={"api_key": "private-agent-key"}, + ) + await http.put( + "/api/v1/preferences/active-models", + json={"default_provider_id": "dashscope", "default_model_id": "qwen-max"}, + ) + ready = await http.get(url, headers={"X-User-Id": "pawapp-scoped"}) + assert ready.json() == {"readiness_version": 1, "model_configured": True} + assert built[-1].api_key == "private-agent-key" + assert "private-agent-key" not in ready.text + + +async def test_analysis_readiness_explicit_model_and_env_fallback(tmp_path, monkeypatch): + async with service_client(tmp_path / "explicit", monkeypatch, model=object()) as (http, _): + assert (await http.get("/api/v1/capabilities/analysis")).json()["model_configured"] + calls = [] + monkeypatch.setattr( + "qwenpaw_data.host.core.api.app.build_model_from_env", + lambda: calls.append(True) or object(), + ) + async with service_client(tmp_path / "env", monkeypatch) as (http, _): + assert (await http.get("/api/v1/capabilities/analysis")).json()["model_configured"] + assert calls == [True] + + async def test_models_and_active_selection(tmp_path, monkeypatch) -> None: async with service_client(tmp_path, monkeypatch, model=object()) as (http, _): added = await http.put( diff --git a/packages/qwenpaw-data-host-core/tests/test_runtime_model_bindings.py b/packages/qwenpaw-data-host-core/tests/test_runtime_model_bindings.py index 95810ab..fecb172 100644 --- a/packages/qwenpaw-data-host-core/tests/test_runtime_model_bindings.py +++ b/packages/qwenpaw-data-host-core/tests/test_runtime_model_bindings.py @@ -63,6 +63,8 @@ async def test_later_turn_uses_new_snapshot_and_keeps_session_agent(monkeypatch) hosts=SimpleNamespace(get=lambda **_: host), ) runtime._executor = SimpleNamespace(start=AsyncMock(), stop=AsyncMock()) + runtime._artifact_snapshot = lambda: {} + runtime.report_analysis_progress = AsyncMock() runtime._deliver_biztrace = AsyncMock() runtime._deliver_followup = AsyncMock() runtime._finish = AsyncMock() diff --git a/packages/qwenpaw-data-host-core/tests/test_stream_objects.py b/packages/qwenpaw-data-host-core/tests/test_stream_objects.py index 2bd3ca8..3a83203 100644 --- a/packages/qwenpaw-data-host-core/tests/test_stream_objects.py +++ b/packages/qwenpaw-data-host-core/tests/test_stream_objects.py @@ -92,6 +92,9 @@ def test_parse_biz_event_segment_artifact_followup() -> None: "session_id": "s1", "name": "report.md", "path": "artifacts/report.md", + "media_type": "text/markdown", + "size_bytes": 7, + "digest": "sha256:" + "a" * 64, "created_at": "2026-01-01T00:00:00Z", "updated_at": "2026-01-01T00:00:00Z", }, diff --git a/packages/qwenpaw-data-host-core/tests/test_submissions_api.py b/packages/qwenpaw-data-host-core/tests/test_submissions_api.py new file mode 100644 index 0000000..0bff14e --- /dev/null +++ b/packages/qwenpaw-data-host-core/tests/test_submissions_api.py @@ -0,0 +1,841 @@ +# -*- coding: utf-8 -*- +from __future__ import annotations + +import asyncio +import json +import sys +import textwrap +from contextlib import asynccontextmanager + +import httpx +import pytest +from sqlalchemy import delete, func, select +from sqlalchemy.exc import OperationalError + +from qwenpaw_data.host.core.api.app import create_app +from qwenpaw_data.host.core.api.models.submissions import SubmitRunRequest +from qwenpaw_data.host.core.api.routers import submissions as routes +from qwenpaw_data.host.core.db.tables import ChatRow, SessionRow, SubmissionRow +from qwenpaw_data.host.core.domain.identity import Identity +from qwenpaw_data.host.core.runtime.chat_runtime import ChatRuntime +from qwenpaw_data.host.core.runtime.registry import get_runtime_registry +from qwenpaw_data.host.core.store import submissions as submission_store +from qwenpaw_data.host.core.stream.output_stream import OutputStream + +PAYLOAD = {"submission_id": "sub_test", "text": "分析收入", "datasource_id": "ds1"} +URL = "/api/v1/submissions" +ANSWER = { + "protocol_version": 1, + "command_id": "cmd_answer_1", + "kind": "answer", + "request_id": "clarification_1", + "answers": [ + { + "question": "Which period?", + "selected_options": ["Q1"], + } + ], +} + + +@pytest.fixture(autouse=True) +def isolated_environment(monkeypatch): + for name in ("QWENPAW_DATA_API_TOKEN", "QWENPAW_DATA_DB_URL", "QWENPAW_DATA_STORE"): + monkeypatch.delenv(name, raising=False) + + +@pytest.fixture +def starts(monkeypatch): + runs = [] + + async def run(self, chat_id, *, identity): + runs.append(chat_id) + + monkeypatch.setattr(ChatRuntime, "run", run) + return runs + + +@asynccontextmanager +async def service(home, **transport_options): + app = create_app(home=home, model=object()) + async with app.router.lifespan_context(app): + async with httpx.AsyncClient( + transport=httpx.ASGITransport( + app=app, + client=("127.0.0.1", 1234), + **transport_options, + ), + base_url="http://testserver", + ) as http: + yield http, app.state.service + + +async def counts(state): + async with state.submissions._sessions() as db: + return [ + await db.scalar(select(func.count()).select_from(table)) + for table in (SubmissionRow, SessionRow, ChatRow) + ] + + +async def accept_without_scheduling(state): + body = SubmitRunRequest(**PAYLOAD) + record, created = await state.submissions.submit( + identity=Identity.anonymous(), + submission_id=body.submission_id, + request_digest=body.request_digest(), + text=body.text, + datasource_id=body.datasource_id, + agent_id=body.agent_id, + ) + assert created + return record + + +async def finish(state, run, status="completed"): + chat = await state.chats.get(run["run_id"]) + stream = OutputStream( + state.events, + session_id=chat.session_id, + chat_id=chat.id, + identity=chat.identity, + ) + await stream.text_end(msg_id="result", index=0, text="Revenue is 42.") + chat.mark_status( + {"completed": "completed", "failed": "failed", "cancelled": "canceled"}[status] + ) + await state.chats.save(chat) + await stream.response(status) + + +def sse_payloads(response): + return [ + json.loads(line[6:]) + for line in response.text.splitlines() + if line.startswith("data: ") + ] + + +async def test_concurrent_retries_create_and_start_exactly_one_run(tmp_path, starts): + async with service(tmp_path) as (http, state): + responses = await asyncio.gather( + *[http.post(URL, json=PAYLOAD) for _ in range(12)] + ) + assert {r.status_code for r in responses} == {202} + runs = [r.json()["run"] for r in responses] + assert len({r["session_id"] for r in runs}) == 1 + assert len({r["run_id"] for r in runs}) == 1 + assert starts == [runs[0]["run_id"]] + assert await counts(state) == [1, 1, 1] + lookup = (await http.get(f"{URL}/sub_test")).json() + assert lookup["state"] == "accepted" + assert lookup["run"]["run_id"] == starts[0] + + +@pytest.mark.parametrize("datasource_id", ["ds1", None]) +async def test_existing_session_turns_preserve_context_and_identity( + tmp_path, starts, datasource_id +): + async with service(tmp_path) as (http, state): + headers = {"X-User-Id": "alice"} + response = await http.post( + "/api/v1/sessions", json={"title": "March GAAP"}, headers=headers + ) + session_id = response.json()["session"]["id"] + readiness_url = f"/api/v1/sessions/{session_id}/submission-readiness" + assert (await http.get(readiness_url, headers=headers)).json() == { + "ready": True + } + assert ( + await http.get(readiness_url, headers={"X-User-Id": "bob"}) + ).status_code == 404 + body = { + **PAYLOAD, + "datasource_id": datasource_id, + "session_id": session_id, + "artifact_comments": [ + { + "path": "report.md", + "line_start": 1, + "line_end": 2, + "comment": "Explain this", + }, + ], + } + accepted = await http.post(URL, json=body, headers=headers) + assert accepted.status_code == 202, accepted.text + run = accepted.json()["run"] + assert (await http.get(readiness_url, headers=headers)).json() == { + "ready": False + } + assert run["session_id"] == session_id + chat = await state.chats.get(run["run_id"]) + assert chat.artifact_comments[0]["comment"] == "Explain this" + assert chat.datasource_id == datasource_id + assert (await http.post(URL, json=body, headers=headers)).json()["run"] == run + assert await counts(state) == [1, 1, 1] + assert ( + await http.post( + URL, json={**body, "submission_id": "another"}, headers=headers + ) + ).status_code == 409 + denied = await http.post(URL, json=body, headers={"X-User-Id": "bob"}) + assert denied.status_code == 404 + assert ( + await http.get( + f"/api/v1/sessions/{session_id}/chats/{run['run_id']}", + headers={"X-User-Id": "bob"}, + ) + ).status_code == 404 + await finish(state, run) + second = await http.post( + URL, json={**body, "submission_id": "next"}, headers=headers + ) + assert second.status_code == 202, second.text + assert second.json()["run"]["session_id"] == session_id + assert await counts(state) == [2, 1, 2] + + +async def test_submission_passes_capability_bridge_to_runtime(tmp_path, monkeypatch): + received = [] + + async def run(self, chat_id, *, identity, capability_bridge=None): + received.append((chat_id, capability_bridge)) + + monkeypatch.setattr(ChatRuntime, "run", run) + bridge = { + "protocol_version": 1, + "endpoint": "http://127.0.0.1:8088/api/pawapp-capabilities", + "token": "x" * 32, + } + async with service(tmp_path) as (http, _state): + response = await http.post(URL, json={**PAYLOAD, "capability_bridge": bridge}) + + assert response.status_code == 202 + assert received == [(response.json()["run"]["run_id"], bridge)] + + +@pytest.mark.parametrize( + "change", + [ + {"text": "different"}, + {"datasource_id": "ds2"}, + {"agent_id": "another"}, + ], +) +async def test_reuse_with_changed_inputs_is_conflict(tmp_path, starts, change): + async with service(tmp_path) as (http, state): + await http.post(URL, json=PAYLOAD) + conflict = await http.post(URL, json={**PAYLOAD, **change}) + assert conflict.status_code == 409 + assert await counts(state) == [1, 1, 1] + assert len(starts) == 1 + + +async def test_separate_submissions_and_identity_namespaces(tmp_path, starts): + async with service(tmp_path) as (http, state): + first = (await http.post(URL, json=PAYLOAD)).json()["run"] + unknown = await http.get(f"{URL}/sub_test", headers={"X-User-Id": "other"}) + assert unknown.json()["state"] == "not_found" + assert ( + await http.get(f"{URL}/sub_test/events", headers={"X-User-Id": "other"}) + ).status_code == 404 + second = ( + await http.post(URL, json=PAYLOAD, headers={"X-User-Id": "other"}) + ).json()["run"] + third = ( + await http.post(URL, json={**PAYLOAD, "submission_id": "sub_other"}) + ).json()["run"] + assert len({r["session_id"] for r in (first, second, third)}) == 3 + assert len(starts) == 3 + assert await counts(state) == [3, 3, 3] + + +async def test_fault_before_commit_rolls_back_all_rows(tmp_path, starts, monkeypatch): + async with service(tmp_path, raise_app_exceptions=False) as (http, state): + original = submission_store._chat_to_row + + def fail(_chat): + raise RuntimeError("injected failure after receipt flush") + + monkeypatch.setattr(submission_store, "_chat_to_row", fail) + assert (await http.post(URL, json=PAYLOAD)).status_code == 500 + assert await counts(state) == [0, 0, 0] + assert (await http.get(f"{URL}/sub_test")).json()["state"] == "not_found" + assert not starts + monkeypatch.setattr(submission_store, "_chat_to_row", original) + assert (await http.post(URL, json=PAYLOAD)).status_code == 202 + assert len(starts) == 1 + + +async def test_lost_http_response_reconciles_original_run( + tmp_path, starts, monkeypatch +): + async with service(tmp_path, raise_app_exceptions=False) as (http, state): + original = routes._lookup_response + + async def fail(*_args): + raise ConnectionError("response lost after acceptance") + + monkeypatch.setattr(routes, "_lookup_response", fail) + assert (await http.post(URL, json=PAYLOAD)).status_code == 500 + monkeypatch.setattr(routes, "_lookup_response", original) + lookup = (await http.get(f"{URL}/sub_test")).json() + retried = (await http.post(URL, json=PAYLOAD)).json() + assert lookup == retried + assert starts == [lookup["run"]["run_id"]] + assert await counts(state) == [1, 1, 1] + + +async def test_http_cancellation_cannot_cancel_acceptance( + tmp_path, starts, monkeypatch +): + async with service(tmp_path) as (http, state): + committed = asyncio.Event() + release = asyncio.Event() + original = state.submissions.submit + + async def pause(**kwargs): + accepted = await original(**kwargs) + committed.set() + await release.wait() + return accepted + + monkeypatch.setattr(state.submissions, "submit", pause) + request = asyncio.create_task( + routes.submit_run( + SubmitRunRequest(**PAYLOAD), + Identity.anonymous(), + state, + ) + ) + await asyncio.wait_for(committed.wait(), 5) + request.cancel() + with pytest.raises(asyncio.CancelledError): + await request + release.set() + await asyncio.gather(*list(state.tasks)) + assert len(starts) == 1 + assert (await http.get(f"{URL}/sub_test")).json()["run"]["run_id"] == starts[0] + + +async def test_answer_command_is_durable_idempotent_and_scoped(tmp_path, starts): + class Runtime: + def __init__(self): + self.answers = [] + + def answer(self, *, clarification_id, result): + self.answers.append((clarification_id, result)) + + async with service(tmp_path) as (http, _state): + run = (await http.post(URL, json=PAYLOAD)).json()["run"] + runtime = Runtime() + get_runtime_registry().register(run["run_id"], runtime) # type: ignore[arg-type] + try: + first = await http.post(f"{URL}/sub_test/commands", json=ANSWER) + retry = await http.post(f"{URL}/sub_test/commands", json=ANSWER) + lookup = await http.get(f"{URL}/sub_test/commands/{ANSWER['command_id']}") + assert first.status_code == 200 + assert first.json()["state"] == "accepted" + assert retry.json() == first.json() == lookup.json() + assert runtime.answers == [ + ( + "clarification_1", + { + "status": "answered", + "answers": ANSWER["answers"], + }, + ) + ] + conflict = await http.post( + f"{URL}/sub_test/commands", + json={ + **ANSWER, + "answers": [{**ANSWER["answers"][0], "selected_options": ["Q2"]}], + }, + ) + assert conflict.status_code == 409 + assert ( + await http.get( + f"{URL}/sub_test/commands/{ANSWER['command_id']}", + headers={"X-User-Id": "other"}, + ) + ).status_code == 404 + finally: + get_runtime_registry().unregister(run["run_id"]) + + +async def test_stale_answer_is_a_stable_rejected_receipt(tmp_path, starts): + async with service(tmp_path) as (http, _state): + await http.post(URL, json=PAYLOAD) + first = await http.post(f"{URL}/sub_test/commands", json=ANSWER) + retry = await http.post(f"{URL}/sub_test/commands", json=ANSWER) + assert first.json()["state"] == "rejected" + assert first.json()["reason"] == "stale_request" + assert retry.json() == first.json() + + +async def test_cancel_command_targets_original_run_once(tmp_path, starts): + class Runtime: + def __init__(self): + self.cancels = 0 + + async def cancel(self): + self.cancels += 1 + + body = { + "protocol_version": 1, + "command_id": "cmd_cancel_1", + "kind": "cancel", + "reason": "user_requested", + } + async with service(tmp_path) as (http, _state): + run = (await http.post(URL, json=PAYLOAD)).json()["run"] + runtime = Runtime() + get_runtime_registry().register(run["run_id"], runtime) # type: ignore[arg-type] + try: + first = await http.post(f"{URL}/sub_test/commands", json=body) + retry = await http.post(f"{URL}/sub_test/commands", json=body) + assert first.json()["state"] == "accepted" + assert retry.json() == first.json() + assert runtime.cancels == 1 + finally: + get_runtime_registry().unregister(run["run_id"]) + + +async def test_prepared_command_queries_as_unknown_after_restart_window( + tmp_path, starts +): + async with service(tmp_path) as (http, state): + record = await accept_without_scheduling(state) + command, created = await state.submissions.prepare_command( + record=record, + command_id=ANSWER["command_id"], + kind="answer", + request_digest="a" * 64, + ) + assert created and command.state == "unknown" + lookup = await http.get(f"{URL}/sub_test/commands/{ANSWER['command_id']}") + assert lookup.json()["state"] == "unknown" + + +async def test_restart_after_commit_before_scheduling_is_interrupted(tmp_path, starts): + async with service(tmp_path) as (_http, state): + record = await accept_without_scheduling(state) + assert not starts + async with service(tmp_path) as (http, state): + lookup = (await http.get(f"{URL}/sub_test")).json() + assert lookup["run"]["run_id"] == record.run_id + assert lookup["run"]["status"] == "interrupted" + assert lookup["run"]["error"]["details"]["reason"] == "executor_restarted" + assert (await http.post(URL, json=PAYLOAD)).json() == lookup + assert not starts + events = sse_payloads(await http.get(f"{URL}/sub_test/events")) + assert len(events) == 1 + assert events[0]["status"] == "cancelled" + # Recovery itself must be idempotent. + async with service(tmp_path) as (http, _state): + assert len(sse_payloads(await http.get(f"{URL}/sub_test/events"))) == 1 + + +async def test_process_exit_after_commit_preserves_receipt(tmp_path, starts): + # A real abrupt exit: no lifespan teardown, runtime cancellation or engine + # disposal can accidentally make this recovery test pass. + script = textwrap.dedent(""" + import asyncio, json, os, sys + from pathlib import Path + from qwenpaw_data.host.core.api.models.submissions import SubmitRunRequest + from qwenpaw_data.host.core.db.engine import create_engine_and_factory, init_db, resolve_db_url + from qwenpaw_data.host.core.domain.identity import Identity + from qwenpaw_data.host.core.store.submissions import SQLSubmissionStore + + async def main(): + engine, factory = create_engine_and_factory(resolve_db_url(Path(sys.argv[1]))) + await init_db(engine) + body = SubmitRunRequest(**json.loads(sys.argv[2])) + record, _ = await SQLSubmissionStore(factory).submit( + identity=Identity.anonymous(), submission_id=body.submission_id, + request_digest=body.request_digest(), text=body.text, + datasource_id=body.datasource_id, agent_id=body.agent_id, + ) + print(record.run_id, flush=True) + os._exit(23) + asyncio.run(main()) + """) + process = await asyncio.create_subprocess_exec( + sys.executable, + "-c", + script, + str(tmp_path), + json.dumps(PAYLOAD), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, stderr = await asyncio.wait_for(process.communicate(), 30) + finally: + if process.returncode is None: + process.kill() + await process.wait() + assert process.returncode == 23, stderr.decode() + run_id = stdout.decode().strip() + async with service(tmp_path) as (http, state): + lookup = (await http.get(f"{URL}/sub_test")).json() + assert lookup["state"] == "accepted" + assert lookup["run"]["run_id"] == run_id + assert lookup["run"]["status"] == "interrupted" + assert (await http.post(URL, json=PAYLOAD)).json() == lookup + assert await counts(state) == [1, 1, 1] + assert not starts + + +async def test_disconnect_while_running_replays_only_unconsumed_events( + tmp_path, starts +): + class ConnectedRequest: + async def is_disconnected(self): + return False + + async with service(tmp_path) as (http, state): + run = (await http.post(URL, json=PAYLOAD)).json()["run"] + stream = OutputStream( + state.events, + session_id=run["session_id"], + chat_id=run["run_id"], + identity=Identity.anonymous(), + ) + await stream.response_created() + response = await routes.submission_events( + ConnectedRequest(), + "sub_test", + Identity.anonymous(), + state, + -1, + None, + ) + iterator = response.body_iterator + first = await anext(iterator) + assert "id: 0" in first + await stream.text_delta(msg_id="result", index=0, text="Revenue") + live = await asyncio.wait_for(anext(iterator), 5) + assert "id: 1" in live + await iterator.aclose() # Simulate transport disconnect after cursor 1. + await finish(state, run) + replay = sse_payloads( + await http.get( + f"{URL}/sub_test/events", + headers={"Last-Event-ID": "1"}, + ) + ) + assert [event["sequence_number"] for event in replay] == [2, 3] + assert replay[0]["text"] == "Revenue is 42." + assert replay[1]["status"] == "completed" + assert len(starts) == 1 + + +@pytest.mark.parametrize("status", ["completed", "failed", "cancelled"]) +async def test_terminal_replay_survives_restart_and_cursor(tmp_path, starts, status): + async with service(tmp_path) as (http, state): + run = (await http.post(URL, json=PAYLOAD)).json()["run"] + await finish(state, run, status) + first = sse_payloads(await http.get(f"{URL}/sub_test/events")) + assert len(first) == 2 + async with service(tmp_path) as (http, state): + replay = sse_payloads(await http.get(f"{URL}/sub_test/events")) + assert replay == first + after = sse_payloads( + await http.get( + f"{URL}/sub_test/events?after_sequence_number=-1", + headers={"Last-Event-ID": "0"}, + ) + ) + assert after == first[1:] + assert ( + sse_payloads( + await http.get(f"{URL}/sub_test/events?after_sequence_number=1") + ) + == [] + ) + lookup = (await http.get(f"{URL}/sub_test")).json() + assert ( + lookup["run"]["status"] + == {"completed": "succeeded", "failed": "failed", "cancelled": "cancelled"}[ + status + ] + ) + assert lookup["run"]["last_sequence_number"] == 1 + assert (await http.post(URL, json=PAYLOAD)).json() == lookup + assert len(starts) == 1 + + +async def test_completed_row_without_terminal_event_is_not_success(tmp_path, starts): + async with service(tmp_path) as (http, state): + run = (await http.post(URL, json=PAYLOAD)).json()["run"] + chat = await state.chats.get(run["run_id"]) + chat.mark_status("completed") + await state.chats.save(chat) + assert (await http.get(f"{URL}/sub_test")).json()["run"][ + "status" + ] == "reconciling" + assert sse_payloads(await http.get(f"{URL}/sub_test/events")) == [] + async with service(tmp_path) as (http, state): + lookup = (await http.get(f"{URL}/sub_test")).json() + assert lookup["run"]["status"] == "interrupted" + assert lookup["run"]["error"]["details"]["previous_status"] == "completed" + assert ( + sse_payloads(await http.get(f"{URL}/sub_test/events"))[0]["status"] + == "cancelled" + ) + + +@pytest.mark.parametrize( + "chat_status,event_status", + [ + ("failed", "failed"), + ("canceled", "cancelled"), + ], +) +async def test_restart_repairs_missing_failure_or_cancel_event( + tmp_path, + starts, + chat_status, + event_status, +): + async with service(tmp_path) as (http, state): + run = (await http.post(URL, json=PAYLOAD)).json()["run"] + chat = await state.chats.get(run["run_id"]) + chat.mark_status(chat_status) + chat.error = {"code": "VALIDATION", "message": "original error"} + await state.chats.save(chat) + async with service(tmp_path) as (http, _state): + event = sse_payloads(await http.get(f"{URL}/sub_test/events"))[0] + assert event["status"] == event_status + assert event["error"]["message"] == "original error" + assert (await http.get(f"{URL}/sub_test")).json()["run"][ + "status" + ] == event_status + + +async def test_runtime_preparation_failure_has_queryable_terminal_result( + tmp_path, monkeypatch +): + # Exercise the real ChatRuntime.run/_finish path without accessing providers. + async with service(tmp_path) as (http, state): + + def not_ready(**kwargs): + raise ValueError("model is not configured") + + monkeypatch.setattr(state.hosts, "get", not_ready) + accepted = await http.post(URL, json=PAYLOAD) + assert accepted.status_code == 202 + await asyncio.gather(*list(state.tasks)) + lookup = (await http.get(f"{URL}/sub_test")).json() + assert lookup["run"]["status"] == "failed" + assert lookup["run"]["error"]["message"] == "model is not configured" + events = sse_payloads(await http.get(f"{URL}/sub_test/events")) + assert len(events) == 1 + assert events[0]["status"] == "failed" + + +async def test_deleted_run_never_becomes_not_found_or_resubmitted(tmp_path, starts): + async with service(tmp_path) as (http, state): + run = (await http.post(URL, json=PAYLOAD)).json()["run"] + await finish(state, run) + assert ( + await http.delete(f"/api/v1/sessions/{run['session_id']}") + ).status_code == 204 + assert (await http.post(URL, json=PAYLOAD)).json()["run"]["run_id"] == run[ + "run_id" + ] + async with state.submissions._sessions() as db, db.begin(): + await db.execute(delete(ChatRow).where(ChatRow.id == run["run_id"])) + lookup = (await http.get(f"{URL}/sub_test")).json() + assert lookup["state"] == "accepted" + assert lookup["run"]["status"] == "unknown" + assert lookup["run"]["reason"] == "run_missing" + assert (await http.post(URL, json=PAYLOAD)).json() == lookup + assert (await http.get(f"{URL}/sub_test/events")).status_code == 410 + assert len(starts) == 1 + + +async def test_database_failure_is_not_authoritative_not_found(tmp_path, monkeypatch): + async with service(tmp_path, raise_app_exceptions=False) as (http, state): + + async def fail(*args): + raise OperationalError("SELECT", {}, Exception("database unavailable")) + + monkeypatch.setattr(state.submissions, "lookup", fail) + assert (await http.get(f"{URL}/sub_test")).status_code == 500 + + +async def test_json_store_explicitly_declines_protocol(tmp_path, monkeypatch): + monkeypatch.setenv("QWENPAW_DATA_STORE", "json") + async with service(tmp_path) as (http, _state): + caps = (await http.get("/api/v1/capabilities/submissions")).json() + assert caps == { + "protocol_version": None, + "durable_submissions": False, + "event_replay": False, + "durable_commands": False, + "scoped_host_capabilities": False, + "artifact_handoff": False, + "analysis_experience": False, + "session_submissions": False, + "optional_session_datasource": False, + "dataset_context": False, + "runtime_model_bindings": False, + "scoped_sessions": True, + } + for response in ( + await http.post(URL, json=PAYLOAD), + await http.get(f"{URL}/sub_test"), + await http.get(f"{URL}/sub_test/events"), + ): + assert response.status_code == 501 + assert ( + response.json()["details"]["reason"] + == "durable_submissions_unsupported" + ) + + +async def test_submission_api_requires_service_bearer(tmp_path, starts, monkeypatch): + monkeypatch.setenv("QWENPAW_DATA_API_TOKEN", "test-token") + async with service(tmp_path) as (http, _state): + assert (await http.post(URL, json=PAYLOAD)).status_code == 401 + assert (await http.get("/api/v1/capabilities/submissions")).status_code == 401 + assert (await http.get(f"{URL}/sub_test")).status_code == 401 + assert (await http.get(f"{URL}/sub_test/events")).status_code == 401 + headers = {"Authorization": "Bearer test-token"} + assert (await http.post(URL, json=PAYLOAD, headers=headers)).status_code == 202 + assert ( + await http.get("/api/v1/capabilities/submissions", headers=headers) + ).json()["protocol_version"] == 1 + + +@pytest.mark.parametrize( + "change", + [ + {"submission_id": ""}, + {"submission_id": "bad/id"}, + {"text": " "}, + {"datasource_id": " "}, + {"protocol_version": 2}, + {"extra": "ignored?"}, + {"agent_id": "../unsafe"}, + ], +) +async def test_invalid_requests_do_not_create_runs(tmp_path, starts, change): + async with service(tmp_path) as (http, state): + response = await http.post(URL, json={**PAYLOAD, **change}) + assert response.status_code in (400, 422) + assert await counts(state) == [0, 0, 0] + assert not starts + + +async def test_host_model_snapshot_is_transient_and_replay_keeps_original( + tmp_path, monkeypatch +): + snapshots = [] + + async def run(self, chat_id, *, identity, runtime_config, context_model_header): + snapshots.append((runtime_config, context_model_header)) + + monkeypatch.setattr(ChatRuntime, "run", run) + binding = { + "provider_id": "openai", + "model_id": "model-a", + "api_key": "runtime-only-key", + "base_url": "https://models.example/v1", + } + payload = { + **PAYLOAD, + "runtime_models": { + "analysis": binding, + "lightweight": binding, + "databridge": binding, + }, + } + body = SubmitRunRequest(**payload) + assert "runtime-only-key" not in repr(body) + assert "runtime_models" not in body.model_dump() + assert body.request_digest() == SubmitRunRequest(**PAYLOAD).request_digest() + async with service(tmp_path) as (http, state): + accepted = await http.post(URL, json=payload) + assert accepted.status_code == 202 + await asyncio.sleep(0) + payload["runtime_models"]["analysis"] = { + **binding, + "model_id": "model-b", + "api_key": "rotated", + } + replay = await http.post(URL, json=payload) + assert replay.json() == accepted.json() + assert len(snapshots) == 1 + assert snapshots[0][0].default.model_id == "model-a" + assert snapshots[0][0].default.api_key == "runtime-only-key" + assert "runtime-only-key" not in repr(snapshots[0][0]) + for path in tmp_path.rglob("*"): + if path.is_file(): + assert b"runtime-only-key" not in path.read_bytes(), str(path) + + +@pytest.mark.parametrize("datasource", [None, "", " "]) +def test_new_submissions_still_require_a_datasource(datasource): + with pytest.raises(ValueError): + SubmitRunRequest(**{**PAYLOAD, "datasource_id": datasource}) + + +def test_optional_session_datasource_has_stable_digest(): + omitted = SubmitRunRequest( + submission_id="s1", text="check model", session_id="ses1" + ) + explicit = SubmitRunRequest( + submission_id="s1", text="check model", session_id="ses1", datasource_id=None + ) + assert omitted.request_digest() == explicit.request_digest() + + +async def test_dataset_context_is_durable_and_idempotent(tmp_path, starts): + from qwenpaw_data.host.core.runtime.turn import TurnInput, compose_agent_input + + payload = {**PAYLOAD, "dataset_names": ["dws_gaap_di"]} + async with service(tmp_path) as (http, state): + response = await http.post(URL, json=payload) + assert response.status_code == 202, response.text + run = response.json()["run"] + chat = await state.chats.get(run["run_id"]) + assert chat.dataset_names == ["dws_gaap_di"] + assert chat.user_input == PAYLOAD["text"] + turn = TurnInput.from_chat(chat) + prompt = compose_agent_input( + turn.user_input, [], dataset_names=turn.dataset_names + ) + assert "dws_gaap_di" in prompt + assert "does not grant access permissions" in prompt + assert (await http.post(URL, json=payload)).json()["run"]["run_id"] == run[ + "run_id" + ] + changed = await http.post(URL, json={**payload, "dataset_names": ["other"]}) + assert changed.status_code == 409 + async with service(tmp_path) as (_, state): + assert (await state.chats.get(run["run_id"])).dataset_names == ["dws_gaap_di"] + + +@pytest.mark.parametrize("names", [[" "], ["x" * 257], ["x"] * 65, [1]]) +def test_dataset_context_rejects_invalid_names(names): + with pytest.raises(ValueError): + SubmitRunRequest(**PAYLOAD, dataset_names=names) + + +def test_dataset_context_empty_preserves_legacy_digest(): + legacy = dict(PAYLOAD) + legacy.pop("submission_id") + legacy.update(protocol_version=1, agent_id="default", capability_bridge=None) + import hashlib + + digest = hashlib.sha256( + json.dumps( + legacy, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode() + ).hexdigest() + assert SubmitRunRequest(**PAYLOAD).request_digest() == digest diff --git a/packages/qwenpaw-data-host-core/tests/test_task_experience_tools.py b/packages/qwenpaw-data-host-core/tests/test_task_experience_tools.py new file mode 100644 index 0000000..bc117f2 --- /dev/null +++ b/packages/qwenpaw-data-host-core/tests/test_task_experience_tools.py @@ -0,0 +1,25 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from qwenpaw_data.host.core.agent.task_experience import build_task_experience_tools + + +async def test_experience_tools_resolve_current_runtime_per_call(): + first = SimpleNamespace( + report_analysis_progress=AsyncMock(), publish_analysis_artifact=AsyncMock() + ) + second = SimpleNamespace( + report_analysis_progress=AsyncMock(), publish_analysis_artifact=AsyncMock() + ) + context = {"analysis_experience": first} + tools = {tool.name: tool for tool in build_task_experience_tools(lambda: context)} + await tools["report_analysis_progress"](stage="analyze") + first.report_analysis_progress.assert_awaited_once_with("analyze") + context["analysis_experience"] = second + await tools["publish_analysis_artifact"]( + path="march.html", role="primary", kind="data/report" + ) + second.publish_analysis_artifact.assert_awaited_once_with( + "march.html", "primary", "data/report", "chat" + ) + first.publish_analysis_artifact.assert_not_awaited() From 5e1aee5847185dde1b12c54ce17bdc448b2da8cb Mon Sep 17 00:00:00 2001 From: Cyrus Zhang Date: Fri, 16 Oct 2026 16:26:36 -0700 Subject: [PATCH 2/2] fix(engine): preserve runtime skill paths and align SSE recovery test Signed-off-by: Cyrus Zhang --- .../host/core/agent/host_capabilities.py | 10 ++++--- .../tests/test_host_capabilities.py | 27 +++++++++++++++---- .../tests/test_submissions_api.py | 25 +++++++++++++++-- 3 files changed, 52 insertions(+), 10 deletions(-) diff --git a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/host_capabilities.py b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/host_capabilities.py index 5a79f8a..25ef37b 100644 --- a/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/host_capabilities.py +++ b/packages/qwenpaw-data-host-core/src/qwenpaw_data/host/core/agent/host_capabilities.py @@ -11,7 +11,7 @@ import shutil import tempfile import threading -from pathlib import Path, PurePosixPath +from pathlib import Path, PurePath, PurePosixPath from typing import Any from urllib.parse import quote @@ -21,6 +21,8 @@ from agentscope.skill import Skill from agentscope.tool import FunctionTool, ToolChunk +from ..artifact_paths import model_visible_path + _MAX_SKILL_BYTES = 2 * 1024 * 1024 _MAX_SKILL_FILES = 1024 _VALID_STATES = {item.value: item for item in ToolResultState} @@ -43,7 +45,7 @@ def __init__( config: dict[str, Any], *, storage_workspace_dir: str | Path, - runtime_workspace_dir: str | Path, + runtime_workspace_dir: str | PurePath, transport: httpx.AsyncBaseTransport | None = None, ) -> None: if config.get("protocol_version") != 1: @@ -66,7 +68,9 @@ def __init__( ): raise HostCapabilityError("invalid_host_capability_bridge") self._storage_root = Path(storage_workspace_dir) / ".pawapp-capabilities" - self._runtime_root = Path(runtime_workspace_dir) / ".pawapp-capabilities" + self._runtime_root = ( + model_visible_path(runtime_workspace_dir) / ".pawapp-capabilities" + ) self._client = httpx.AsyncClient( base_url=str(url).rstrip("/") + "/", headers={"Authorization": f"Bearer {token}"}, diff --git a/packages/qwenpaw-data-host-core/tests/test_host_capabilities.py b/packages/qwenpaw-data-host-core/tests/test_host_capabilities.py index a8bdcea..069f410 100644 --- a/packages/qwenpaw-data-host-core/tests/test_host_capabilities.py +++ b/packages/qwenpaw-data-host-core/tests/test_host_capabilities.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +from pathlib import PureWindowsPath import httpx import pytest @@ -45,7 +46,9 @@ def descriptor( } -def client(tmp_path, handler) -> HostCapabilityClient: +def client( + tmp_path, handler, *, runtime_workspace_dir="/workspace", +) -> HostCapabilityClient: return HostCapabilityClient( { "protocol_version": 1, @@ -53,12 +56,25 @@ def client(tmp_path, handler) -> HostCapabilityClient: "token": "scoped-token", }, storage_workspace_dir=tmp_path, - runtime_workspace_dir="/workspace", + runtime_workspace_dir=runtime_workspace_dir, transport=httpx.MockTransport(handler), ) -async def test_remote_tool_and_skill_are_loaded_with_scoped_auth(tmp_path) -> None: +@pytest.mark.parametrize( + ("runtime_workspace_dir", "expected_runtime_root"), + [ + ("/workspace", "/workspace/.pawapp-capabilities"), + (PureWindowsPath("/workspace"), "/workspace/.pawapp-capabilities"), + (r"C:\workspace", r"C:\workspace\.pawapp-capabilities"), + (r"\\server\share\workspace", r"\\server\share\workspace\.pawapp-capabilities"), + ], +) +async def test_remote_tool_and_skill_are_loaded_with_scoped_auth( + tmp_path, + runtime_workspace_dir, + expected_runtime_root, +) -> None: calls: list[tuple[str, str]] = [] tool = descriptor( "app/tool/echo", @@ -118,7 +134,7 @@ def handler(request: httpx.Request) -> httpx.Response: }, ) - bridge = client(tmp_path, handler) + bridge = client(tmp_path, handler, runtime_workspace_dir=runtime_workspace_dir) try: tools = await bridge.build_tools() skills = await bridge.load_skills() @@ -131,7 +147,8 @@ def handler(request: httpx.Request) -> httpx.Response: assert result.state is ToolResultState.SUCCESS assert result.content[0].text == "echo: hello" assert skills[0].name == "guidance" - assert skills[0].dir == f"/workspace/.pawapp-capabilities/{'b' * 64}" + separator = "\\" if "\\" in expected_runtime_root else "/" + assert skills[0].dir == f"{expected_runtime_root}{separator}{'b' * 64}" materialized = tmp_path / ".pawapp-capabilities" / ("b" * 64) assert (materialized / "references/example.txt").read_text() == "example" assert sum(path.endswith("/catalog") for _, path in calls) == 1 diff --git a/packages/qwenpaw-data-host-core/tests/test_submissions_api.py b/packages/qwenpaw-data-host-core/tests/test_submissions_api.py index 0bff14e..cade416 100644 --- a/packages/qwenpaw-data-host-core/tests/test_submissions_api.py +++ b/packages/qwenpaw-data-host-core/tests/test_submissions_api.py @@ -568,7 +568,14 @@ async def test_terminal_replay_survives_restart_and_cursor(tmp_path, starts, sta assert len(starts) == 1 -async def test_completed_row_without_terminal_event_is_not_success(tmp_path, starts): +async def test_completed_row_without_terminal_event_is_not_success( + tmp_path, starts, monkeypatch, +): + class ConnectedRequest: + async def is_disconnected(self): + return False + + monkeypatch.setenv("QWENPAW_DATA_STREAM_SSE_HEARTBEAT_SECONDS", "0.01") async with service(tmp_path) as (http, state): run = (await http.post(URL, json=PAYLOAD)).json()["run"] chat = await state.chats.get(run["run_id"]) @@ -577,7 +584,21 @@ async def test_completed_row_without_terminal_event_is_not_success(tmp_path, sta assert (await http.get(f"{URL}/sub_test")).json()["run"][ "status" ] == "reconciling" - assert sse_payloads(await http.get(f"{URL}/sub_test/events")) == [] + # A saved terminal status can precede the final event. Keep streaming + # until that event arrives, or the client disconnects before restart. + response = await routes.submission_events( + ConnectedRequest(), + "sub_test", + Identity.anonymous(), + state, + -1, + None, + ) + iterator = response.body_iterator + try: + assert await asyncio.wait_for(anext(iterator), 5) == ": keepalive\n\n" + finally: + await iterator.aclose() async with service(tmp_path) as (http, state): lookup = (await http.get(f"{URL}/sub_test")).json() assert lookup["run"]["status"] == "interrupted"