diff --git a/README.md b/README.md index cd6edfa..e79b8b7 100644 --- a/README.md +++ b/README.md @@ -78,15 +78,7 @@ vidxp doctor vidxp mcp-config ``` -`vidxp mcp-config` emits the `mcpServers` JSON used by Claude Desktop and -compatible local stdio clients. Codex has separate configuration: use -`codex mcp add vidxp -- vidxp-mcp --repository default` or configure -`[mcp_servers.vidxp]` in `~/.codex/config.toml`. ChatGPT Desktop and ChatGPT web -use ChatGPT's own connector configuration and do not read Codex's config; hosted -ChatGPT connections use a remote MCP endpoint, with VidXP deployed behind HTTPS -and OIDC. - -The CLI works without MCP. Add the browser app with: +Add the browser interface with: ```bash uv tool install --python 3.14 --torch-backend cpu \ @@ -94,34 +86,17 @@ uv tool install --python 3.14 --torch-backend cpu \ vidxp ui ``` -`vidxp ui` binds to loopback by default. Use `vidxp ui --share` only when you -intend to expose the unauthenticated browser interface on the local network. -Streamlit prints its Local and Network URLs when it starts. VidXP disables -Streamlit's first-run email prompt and usage-statistics collection. - -`vidxp-api --share` exposes a bearer-protected HTTP API/MCP endpoint on a -trusted LAN. Browser upload tools are omitted unless an explicit HTTPS upload -handoff origin is configured, because a LAN listener alone cannot advertise a -safe browser capability page. - -If the `vidxp` command is not found, run `uv tool update-shell` once and reopen -the terminal. +See the [installation guide](INSTALLATION_GUIDE.md) for client-specific MCP +configuration, the HTTP API, and remote server setup. ### 2. Desktop app Download the installer for Windows, Apple Silicon macOS, or Linux from [GitHub Releases](https://github.com/grayhatdevelopers/vidxp/releases). -On first launch, choose whether to adopt an existing compatible VidXP -installation without downloading another runtime, or create a private runtime -managed by VidXP Desktop. For a managed runtime, choose the search capabilities, -model location, and optional browser interface; Python and uv do not need to be -installed separately. - -Desktop opens its control panel when started. Browser launch is a separate, -explicit **Open VidXP** action. After configuration, closing the control panel -keeps VidXP available through **Manage VidXP**, **Open VidXP**, and **Quit -VidXP** in the system tray. +Connect an existing VidXP installation or let the desktop app manage an isolated +runtime for you. See the [desktop guide](docs/desktop.md) for supported setup +options. ### 3. Docker for a server @@ -141,21 +116,12 @@ See the [Coolify guide](docs/deployment/coolify.md) for the complete setup. ## What you can do today -- Build a reusable search library from one video or a whole collection. -- Search dialogue by meaning, even when you do not remember the exact words. -- Find visual moments by describing the scene you are looking for. -- Group recurring faces in a video and render a highlighted actor overlay. -- Search one selected video or every video in the active library. -- Open matching timestamps and export downloadable clips and overlays. -- Receive search and query results as annotated evidence boards, then open the - exact frames or clips worth inspecting. -- Retrieve completed clips through native MCP resources, local stdio paths, or - short-lived resumable HTTPS downloads without embedding video bytes in tool JSON. +- Build searchable libraries from individual videos or whole collections. +- Find dialogue by meaning and visual moments by describing the scene. +- Ask grounded questions and inspect the supporting boards, frames, or clips. +- Group recurring faces and render highlighted actor overlays. - Keep personal, client, or project libraries separate. -- Follow long indexing jobs, cancel them, and keep the last working index if a - later run fails. -- Use the browser app, automate the CLI, connect an MCP agent, or integrate - VidXP into another application. +- Use VidXP through the desktop app, browser, CLI, MCP, or HTTP API. ## A first search @@ -188,31 +154,10 @@ application, or use the HTTP API when VidXP runs as a service. [![VidXP being used with ChatGPT Desktop AI](./docs/images/claude-with-vidxp.jpg)](https://youtu.be/fa4Zx-bSOh4) -MCP clients can add and discover videos, start indexing, search dialogue and -scenes, ask questions about a library, and create clips or actor overlays. -Local agents can connect over stdio; remote agents can connect to a -self-hosted VidXP server. - -Search and question results can include directly inspectable frames and clips, so -agents can show the evidence behind an answer without making users translate raw -timestamps. Evidence rendering is best-effort: a result can still be useful when -an individual frame or clip cannot be produced. Agents can request additional -ranked evidence in small batches without rerunning the search. -For broader result sets, an agent can first show bounded evidence-board pages -with a tile-to-evidence map, then fetch exact frames or clips only for the -selected tiles. - -Remote agents can hand users a short-lived page for selecting and uploading -multiple videos. Local agents can ingest approved filesystem paths without moving -video bytes through MCP. VidXP normally indexes successful imports automatically; -registration-only ingestion stops at `registered`, and indexing failures can be -retried without uploading the video again. - -Agents can call `get_workspace` before acting to inspect registered media, -active-index coverage, model readiness, and the searchable, queryable, -inspectable, or renderable roles available for each video. Invalid capability -or media selections are rejected before a durable job is queued and include an -actionable next step. +MCP lets AI clients add and index videos, search dialogue and scenes, ask +questions about a library, and return inspectable evidence such as boards, +frames, and clips. Clients can connect locally over stdio or to a self-hosted +VidXP server. ### ChatGPT and Codex skills diff --git a/docs/architecture/platform.md b/docs/architecture/platform.md index 6cb0d0a..aecfdbf 100644 --- a/docs/architecture/platform.md +++ b/docs/architecture/platform.md @@ -920,15 +920,22 @@ guidance. MCP search and query jobs compile their ranked candidates into annotated, media-separated JPEG evidence boards before the original job succeeds. The board uses the existing evidence-frame and artifact services, returns native image -resources plus a tile map to stable evidence IDs, and is included in the same -`get_job` response. The default budget is 24 tiles per page and four pages per -job; `next_start_rank` continues without imposing the standalone-artifact limit. +resources plus a concise index to stable evidence IDs. Active observation uses +compact job summaries and bounded waits; after the job becomes terminal, +`get_job_evidence` returns content-only board images, links, timestamps, and +evidence IDs that remain model-visible in hosts that prioritize +`structuredContent` over MCP content blocks. `get_job` is the explicit full +machine record and does not load or duplicate evidence bytes. The default budget +is 24 tiles per page and four pages per job; `next_start_rank` continues without +imposing the standalone-artifact limit. Callers may additionally request `keyframes` or `keyframes_and_clips`, with a hard maximum of five standalone items in the initial job. Scene evidence extracts the authoritative indexed frame number; other intervals and anonymous actor clusters use a labeled representative frame. `materialize_job_evidence` accepts one to ten -board evidence IDs for later frame or clip drill-down without rerunning retrieval. +board evidence IDs for later frame or clip drill-down without rerunning retrieval; +it likewise returns content-only images and links rather than duplicating its full +delivery model. `create_evidence_board` remains for custom selections and continuation pages, and `create_evidence_clip` remains the single-item fallback. Clip rendering requires repository write scope; boards and keyframes retain read scope. @@ -1056,13 +1063,18 @@ Initial curated tools: - `get_artifact_download` - `list_jobs` - `get_job` +- `get_job_evidence` +- `get_job_status` +- `wait_job` - `retry_job` - `cancel_job` Media and job discovery let an agent recover registered assets and durable work -without carrying IDs across sessions. Generic job polling, retry, and cancellation -cover indexing, search, and query without duplicating operation contracts. Video -bytes remain on the HTTP/tus ingestion boundary rather than crossing MCP. +without carrying IDs across sessions. Compact summaries and `wait_job` provide +bounded observation without repeatedly returning completed result payloads; +retry and cancellation cover indexing, search, and query without duplicating +operation contracts. Video bytes remain on the HTTP/tus ingestion boundary +rather than crossing MCP. Tool results use real output schemas and structured content. Descriptions remain short and agent-oriented; full API response schemas are not embedded as prose. diff --git a/docs/deployment/coolify.md b/docs/deployment/coolify.md index 863819d..e07fcbf 100644 --- a/docs/deployment/coolify.md +++ b/docs/deployment/coolify.md @@ -175,17 +175,18 @@ curl --fail-with-body \ --data '{"modalities":["dialogue","scene","actor"],"capability_options":{}}' ``` -The `202 Accepted` response contains the durable `job_id` and a `Location` -header. Poll that location until the job succeeds: +The `202 Accepted` response contains the durable `job_id`. Use the bounded wait +endpoint for compact status, passing the returned observation token on the next +request. Fetch the full job once after it becomes terminal: ```bash curl --fail-with-body \ --header "Authorization: Bearer ${VIDXP_HTTP_STATIC_BEARER_TOKEN}" \ - "https://${VIDXP_PUBLIC_API_HOST}/api/v1/jobs/" + "https://${VIDXP_PUBLIC_API_HOST}/api/v1/jobs//wait?timeout_seconds=30" ``` -The Streamable HTTP MCP `prepare_models` and `get_job` tools expose the same -operation for an authenticated agent client. Check +The Streamable HTTP MCP `prepare_models`, `wait_job`, and `get_job` tools expose +the same operation for an authenticated agent client. Check `/api/v1/runtime/readiness` afterward; `/ready` covers control-plane availability and does not claim that every optional model is prepared. diff --git a/skills/vidxp-find-video-evidence/SKILL.md b/skills/vidxp-find-video-evidence/SKILL.md index e5abd6a..0f8027b 100644 --- a/skills/vidxp-find-video-evidence/SKILL.md +++ b/skills/vidxp-find-video-evidence/SKILL.md @@ -10,18 +10,25 @@ description: Use VidXP to search indexed videos, answer grounded questions about 1. Resolve the `vidxp` MCP tools, then call `get_workspace`. If the requested video is not indexed, explain that it must be indexed first. 2. Use `search_moments` to locate moments or `query_video` for a synthesized, - grounded answer. Set `command.media_id` when the user means one video. + grounded answer. Use `command.query` with `search_moments` and + `command.question` with `query_video`. Set `command.media_id` when the user + means one video. 3. Omit `command.evidence_delivery` for the normal path. The completed job includes an annotated board covering the ranked results. -4. Poll only that job with `get_job`, honoring `poll_after_seconds`. Search and - query may take time; update the user when the stage changes or about once per - minute, never on every poll and never with an invented ETA. +4. Call `wait_job` for bounded waits. Pass its `observation_token` as + `after_observation_token` on the next wait. When terminal, call + `get_job_evidence` once. It returns the concise evidence index and visual + content without the full structured job dump. Use `get_job` only when exact + machine fields not present in that index are actually needed. Search and + query may take time; update the user + when the stage changes or about once per minute, never after every wait and + never with an invented ETA. 5. Inspect and show the returned board before making visual claims. Use its tile evidence IDs for follow-up: - `materialize_job_evidence` accepts up to ten selected IDs and returns - standalone keyframes or clips without rerunning retrieval. + model-visible standalone keyframes or clip links without rerunning retrieval. - `create_evidence_board` is only for a custom selection or the - `next_start_rank` continuation; poll its returned job ID. + `next_start_rank` continuation; wait on its returned job ID the same way. 6. When standalone artifacts are required in the initial job, put exactly this inside `command`: `"evidence_delivery": {"mode": "keyframes_and_clips", "max_items": 3}`. Never send @@ -38,11 +45,12 @@ description: Use VidXP to search indexed videos, answer grounded questions about ## Output -- Present returned board images, frames, clips, or working resource links—not - timestamps alone. If a host does not render a link, use - `get_artifact_download`. +- The final response must visibly embed a returned board or frame, or include a + working downloadable resource link—not timestamps alone. Use the returned + `local_path` or `download_url`; never write an unlinked label such as “View + evidence board.” Use `get_artifact_download` only if neither is returned. - Preserve the source job and evidence IDs. Describe scores as retrieval scores, and distinguish a visible appearance from a dialogue or caption mention. -- Stop polling on success, failure, or cancellation. An empty result means no +- Stop waiting on success, failure, or cancellation. An empty result means no matching indexed evidence was found, not that the event is absent from the original video. diff --git a/skills/vidxp-ingest-video/SKILL.md b/skills/vidxp-ingest-video/SKILL.md index 7fc65d3..198023f 100644 --- a/skills/vidxp-ingest-video/SKILL.md +++ b/skills/vidxp-ingest-video/SKILL.md @@ -13,7 +13,8 @@ description: Use VidXP to upload, import, register, and automatically index vide for ordinary content retrieval. Add `actor` only when anonymous recurring-face clusters are wanted; it does not identify people by name. 3. Call `get_runtime_readiness`. If selected models are missing, submit - `prepare_models` for those modalities and poll its job with `get_job`. + `prepare_models`, use `wait_job` with its observation token for subsequent + bounded waits, then fetch `get_job` once when terminal. 4. Use `ingest_local_media` for one to ten paths accessible to VidXP; otherwise use `create_media_upload` and give the returned link to the user. Keep `index_after_import` enabled unless registration-only behavior was requested. @@ -30,5 +31,5 @@ description: Use VidXP to upload, import, register, and automatically index vide - Tell the user that model preparation and indexing can take several minutes. - Update when the stage changes or about once per minute; do not narrate every - poll or invent an ETA. + status check or invent an ETA. - Treat files independently so one failure does not hide successful siblings. diff --git a/src/vidxp/api_routes/jobs.py b/src/vidxp/api_routes/jobs.py index 96a2666..c6c8e5e 100644 --- a/src/vidxp/api_routes/jobs.py +++ b/src/vidxp/api_routes/jobs.py @@ -15,14 +15,19 @@ CreateActorOverlayCommand, CreateIndexCommand, CreateSnippetCommand, + DEFAULT_JOB_WAIT_SECONDS, Job, JobPage, JobResult, + JobSummary, + JobWaitResult, ListJobsCommand, + MAX_JOB_WAIT_SECONDS, Principal, PrepareModelsCommand, QueryVideoCommand, SearchCommand, + Sha256, ) from vidxp.composition import HttpApplicationContext from vidxp.core.identifiers import JobId @@ -287,6 +292,49 @@ def get_job( return _http_job(service.jobs.get(job_id)) +@router.get( + "/{job_id}/status", + response_model=JobSummary, + operation_id="getJobStatus", + summary="Get compact job status", + dependencies=[Depends(read_principal)], +) +def get_job_status( + job_id: JobId, + service: Annotated[HttpApplicationContext, Depends(context)], +) -> JobSummary: + return service.jobs.summary(job_id) + + +@router.get( + "/{job_id}/wait", + response_model=JobWaitResult, + operation_id="waitForJobChange", + summary="Wait for a job change", + description=( + "Wait up to 30 seconds for a job to change public stage or become " + "terminal. Evidence rendering is treated as one observable phase " + "rather than one change per artifact. Pass the previous observation " + "token on subsequent requests." + ), + dependencies=[Depends(read_principal)], +) +def wait_for_job_change( + job_id: JobId, + service: Annotated[HttpApplicationContext, Depends(context)], + after_observation_token: Sha256 | None = None, + timeout_seconds: Annotated[ + int, + Query(gt=0, le=MAX_JOB_WAIT_SECONDS), + ] = DEFAULT_JOB_WAIT_SECONDS, +) -> JobWaitResult: + return service.jobs.wait_for_change( + job_id, + after=after_observation_token, + timeout_seconds=timeout_seconds, + ) + + @router.get( "/{job_id}/result", response_model=JobResult, diff --git a/src/vidxp/application_models.py b/src/vidxp/application_models.py index b3314f3..d6c64e6 100644 --- a/src/vidxp/application_models.py +++ b/src/vidxp/application_models.py @@ -1303,6 +1303,10 @@ class PrepareModelsResult(ApplicationModel): JOB_SCHEMA_VERSION = 2 JOB_PROGRESS_SCHEMA_VERSION = 1 +JOB_SUMMARY_SCHEMA_VERSION = 1 +JOB_WAIT_RESULT_SCHEMA_VERSION = 1 +DEFAULT_JOB_WAIT_SECONDS = 30 +MAX_JOB_WAIT_SECONDS = 30 class IndexSnapshotReference(ApplicationModel): @@ -1505,6 +1509,35 @@ def _validate_terminal_payload(self) -> "Job": return self +class JobSummary(ApplicationModel): + """Compact observation of a durable job without its typed result payload.""" + + schema_version: Literal[JOB_SUMMARY_SCHEMA_VERSION] = JOB_SUMMARY_SCHEMA_VERSION + job_id: JobId + kind: JobKind + state: JobState + queue: JobQueue + progress: JobProgress | None = None + error: ErrorDetail | None = None + recovery_attempts: int = Field(default=0, ge=0) + created_at: AwareDatetime | None = None + updated_at: AwareDatetime | None = None + terminal: bool + result_available: bool + observation_token: Sha256 + + +class JobWaitResult(ApplicationModel): + """Result of waiting for a meaningful job state or stage change.""" + + schema_version: Literal[JOB_WAIT_RESULT_SCHEMA_VERSION] = ( + JOB_WAIT_RESULT_SCHEMA_VERSION + ) + job: JobSummary + changed: bool + timed_out: bool + + class ListJobsCommand(ApplicationModel): page_size: int = Field(default=50, gt=0, le=100) cursor: str | None = Field(default=None, min_length=1, max_length=512) diff --git a/src/vidxp/job_service.py b/src/vidxp/job_service.py index bb9201a..45e369e 100644 --- a/src/vidxp/job_service.py +++ b/src/vidxp/job_service.py @@ -1,7 +1,9 @@ from __future__ import annotations +import hashlib +import json from functools import wraps -from time import sleep +from time import monotonic, sleep from typing import Any, Callable, Protocol from vidxp.application_models import ( @@ -20,6 +22,10 @@ JobQueue, JobResult, JobState, + JobSummary, + JobWaitResult, + DEFAULT_JOB_WAIT_SECONDS, + MAX_JOB_WAIT_SECONDS, InvalidRequestError, ListJobsCommand, MediaImportJobRequest, @@ -54,6 +60,16 @@ def plan_actor_overlay( ) -> ActorOverlayJobRequest: ... +_EVIDENCE_JOB_KINDS = frozenset( + { + JobKind.search, + JobKind.query, + JobKind.evidence_board, + } +) +_RETRIEVAL_STAGES = frozenset({"querying", "searching"}) + + def job_boundary(handler: Callable) -> Callable: """Translate job-backend failures once for every adapter.""" @@ -256,6 +272,62 @@ def get(self, job_id: str) -> Job: raise ResourceNotFoundError("job") return job + @job_boundary + def summary(self, job_id: str) -> JobSummary: + return self._summary(self.get(job_id)) + + @job_boundary + def wait_for_change( + self, + job_id: str, + *, + after: str | None = None, + timeout_seconds: float = DEFAULT_JOB_WAIT_SECONDS, + ) -> JobWaitResult: + if timeout_seconds <= 0 or timeout_seconds > MAX_JOB_WAIT_SECONDS: + raise InvalidRequestError( + errors=[ + { + "field": "timeout_seconds", + "message": ( + "The timeout must be greater than zero and no more " + f"than {MAX_JOB_WAIT_SECONDS} seconds." + ), + } + ] + ) + initial = self.get(job_id) + initial_summary = self._summary(initial) + if initial_summary.terminal: + return JobWaitResult( + job=initial_summary, + changed=( + after is not None and initial_summary.observation_token != after + ), + timed_out=False, + ) + if after is not None and initial_summary.observation_token != after: + return JobWaitResult( + job=initial_summary, + changed=True, + timed_out=False, + ) + + baseline = after or initial_summary.observation_token + job, timed_out = self._poll_until( + job_id, + deadline=monotonic() + timeout_seconds, + predicate=lambda candidate: ( + candidate.terminal + or self._summary(candidate).observation_token != baseline + ), + ) + return JobWaitResult( + job=self._summary(job), + changed=not timed_out, + timed_out=timed_out, + ) + @job_boundary def list(self, command: ListJobsCommand) -> JobPage: return self.backend.list(command) @@ -350,16 +422,93 @@ def wait( progress: Callable[[Job], None] | None = None, ) -> Job: last_progress = None - while True: - job = self.get(job_id) + + def report(job: Job) -> None: + nonlocal last_progress if progress is not None and job.progress != last_progress: progress(job) last_progress = job.progress - if job.state not in {JobState.queued, JobState.running}: - if job.state != JobState.succeeded: - self.result(job.job_id) - return job - sleep(self.settings.workflow_poll_interval_seconds) + + job, _timed_out = self._poll_until( + job_id, + deadline=None, + predicate=lambda candidate: candidate.terminal, + on_job=report, + ) + if job.state != JobState.succeeded: + self.result(job.job_id) + return job + + def _poll_until( + self, + job_id: str, + *, + deadline: float | None, + predicate: Callable[[Job], bool], + on_job: Callable[[Job], None] | None = None, + ) -> tuple[Job, bool]: + while True: + job = self.get(job_id) + if on_job is not None: + on_job(job) + if predicate(job): + return job, False + if deadline is not None: + remaining = deadline - monotonic() + if remaining <= 0: + return job, True + interval = min( + self.settings.workflow_poll_interval_seconds, + remaining, + ) + else: + interval = self.settings.workflow_poll_interval_seconds + sleep(interval) + + @staticmethod + def _summary(job: Job) -> JobSummary: + progress = job.progress + error = job.error + observation_stage = None if progress is None else progress.stage + if ( + not job.terminal + and job.kind in _EVIDENCE_JOB_KINDS + and observation_stage not in _RETRIEVAL_STAGES + ): + observation_stage = "rendering_evidence" + observation = { + "job_id": job.job_id, + "kind": job.kind.value, + "state": job.state.value, + "queue": job.queue.value, + "stage": observation_stage, + "total": None if progress is None else progress.total, + "error_code": None if error is None else error.code, + "recovery_attempts": job.recovery_attempts, + "terminal": job.terminal, + "result_available": job.result is not None, + } + token = hashlib.sha256( + json.dumps( + observation, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + ).hexdigest() + return JobSummary( + job_id=job.job_id, + kind=job.kind, + state=job.state, + queue=job.queue, + progress=progress, + error=error, + recovery_attempts=job.recovery_attempts, + created_at=job.created_at, + updated_at=job.updated_at, + terminal=job.terminal, + result_available=job.result is not None, + observation_token=token, + ) def _model_queue(self) -> JobQueue: return ( diff --git a/src/vidxp/mcp.py b/src/vidxp/mcp.py index 10cada7..991afb7 100644 --- a/src/vidxp/mcp.py +++ b/src/vidxp/mcp.py @@ -43,10 +43,12 @@ CreateIndexCommand, ErrorCategory, ErrorDetail, + EvidenceDeliveryItem, EvidenceDeliveryMode, EvidenceDeliveryPolicy, EvidenceDeliveryResult, EvidenceBoardResult, + DEFAULT_JOB_WAIT_SECONDS, FusedSearchResult, Identifier, InitialEvidenceDeliveryPolicy, @@ -54,11 +56,14 @@ Job, JobId, JobKind, - JobState, JobPage, + JobState, + JobSummary, + JobWaitResult, ListJobsCommand, ListMediaCommand, LocalMediaIngestionCommand, + MAX_JOB_WAIT_SECONDS, MediaAsset, MediaId, MediaPage, @@ -513,8 +518,8 @@ async def lifecycle(_server): "Call get_workspace before planning index, search, query, or actor " "work; it reports valid capability roles for each media item. Call " "get_runtime_readiness before indexing. If selected model " - "artifacts are missing, submit prepare_models and poll get_job " - "until it completes. " + "artifacts are missing, submit prepare_models and use wait_job " + "until it completes, then fetch the full job once. " f"{ingestion_instructions}" "Automatic indexing uses every indexable capability exposed by " "the repository runtime unless modalities are supplied; set " @@ -526,8 +531,9 @@ async def lifecycle(_server): "snapshot. MCP defaults to an annotated evidence board in the same " "completed search/query job; request keyframes or " "keyframes_and_clips only for standalone drill-down artifacts. The " - "ordinary flow is submit search/query, then poll that job and inspect " - "its board. Use create_evidence_board only for custom selections or " + "ordinary flow is submit search/query, use wait_job for bounded " + "status observation, then call get_job once and " + "inspect its board. Use create_evidence_board only for custom selections or " "continuation pages. Use materialize_job_evidence with evidence " "IDs from the completed result to inspect additional candidates in " "batches of ten without rerunning retrieval or supplying timestamps. " @@ -820,9 +826,7 @@ async def project_evidence_delivery( projected_board = None blocks: list[ImageContent | ResourceLink | TextContent] = [] if delivery.board is not None: - projected_board, board_blocks = await project_evidence_board( - delivery.board - ) + projected_board, board_blocks = await project_evidence_board(delivery.board) blocks.extend(board_blocks) projected_items = [] for item in delivery.items: @@ -951,6 +955,146 @@ async def project_evidence_board( blocks.append(link) return board.model_copy(update={"pages": tuple(projected_pages)}), blocks + def concise_text(value: str | None, *, limit: int = 320) -> str | None: + if value is None: + return None + compact = " ".join(value.split()) + if len(compact) <= limit: + return compact + return f"{compact[: limit - 1].rstrip()}…" + + def presentation_target(delivery: ArtifactDownload | None) -> str | None: + if delivery is None: + return None + if delivery.local_path is not None: + return f"local_path={delivery.local_path}" + if delivery.download_url is not None: + return f"download_url={delivery.download_url}" + if delivery.resource_uri is not None: + return f"resource_uri={delivery.resource_uri}" + return None + + def presentation_artifacts(delivery: EvidenceDeliveryResult) -> list[str]: + artifacts: list[str] = [] + if delivery.board is not None: + for page in delivery.board.pages: + target = presentation_target(page.artifact.delivery) + if target is not None: + artifacts.append(f"- board page {page.page_number} | {target}") + for item in delivery.items: + if item.keyframe is not None: + target = presentation_target(item.keyframe.artifact.delivery) + if target is not None: + artifacts.append(f"- evidence {item.evidence_id} frame | {target}") + if item.clip is not None: + target = presentation_target(item.clip.delivery) + if target is not None: + artifacts.append(f"- evidence {item.evidence_id} clip | {target}") + return artifacts + + def evidence_index( + *, + source_job_id: JobId, + delivery: EvidenceDeliveryResult, + query_result: QueryAnswer | None = None, + ) -> str: + lines = [f"VidXP evidence for job {source_job_id}."] + if query_result is not None: + if query_result.claims: + lines.append("Grounded answer:") + for claim in query_result.claims: + text = concise_text(claim.text, limit=512) or "" + citations = ", ".join(claim.evidence_ids) + lines.append(f"- {text} [evidence: {citations}]") + elif query_result.fallback_reason: + lines.append( + "Answer unavailable: " + + (concise_text(query_result.fallback_reason, limit=512) or "") + ) + + board = delivery.board + if board is not None: + lines.append( + f"Board: {board.rendered_count}/{board.requested_count} candidates " + f"rendered across {len(board.pages)} page(s); " + f"{board.failed_count} failed." + ) + candidates = board.tiles + else: + candidates = delivery.items + + if candidates: + lines.append( + "Evidence index (rank | source seconds | modalities | ID | label):" + ) + for candidate in candidates: + if isinstance(candidate, EvidenceDeliveryItem): + resolved = candidate.range + start = resolved.source_start_seconds if resolved is not None else 0.0 + end = resolved.source_end_seconds if resolved is not None else start + label = None + else: + start = candidate.start + end = candidate.end + label = concise_text(candidate.display_text) + line = ( + f"- {candidate.rank} | {start:.3f}-{end:.3f} | " + f"{','.join(candidate.modalities)} | {candidate.evidence_id}" + ) + if label: + line += f" | {label}" + lines.append(line) + + artifacts = presentation_artifacts(delivery) + if artifacts: + lines.append( + "User-presentable artifacts (embed local_path or link download_url; " + "do not invent an unlinked evidence label):" + ) + lines.extend(artifacts) + + if board is not None and board.next_start_rank is not None: + lines.append(f"More candidates start at rank {board.next_start_rank}.") + lines.append( + "Use materialize_job_evidence with this job ID and up to ten evidence " + "IDs for standalone frames or clips." + ) + return "\n".join(lines) + + async def evidence_content( + job: Job, + ) -> list[ImageContent | ResourceLink | TextContent]: + query_result = None + if job.kind in {JobKind.search, JobKind.query}: + result = job.result.result + delivery = result.evidence_delivery + if delivery is None: + raise ApplicationError( + "job_evidence_unavailable", + ErrorCategory.conflict, + "The completed job does not contain deliverable evidence.", + ) + if job.kind == JobKind.query: + query_result = result + projected_delivery, blocks = await project_evidence_delivery(delivery) + index = evidence_index( + source_job_id=job.job_id, + delivery=projected_delivery, + query_result=query_result, + ) + else: + board = job.result.result + projected_board, blocks = await project_evidence_board(board) + index = evidence_index( + source_job_id=board.source_job_id, + delivery=EvidenceDeliveryResult( + policy=EvidenceDeliveryPolicy(mode=EvidenceDeliveryMode.none), + items=(), + board=projected_board, + ), + ) + return [TextContent(type="text", text=index), *blocks] + def completed_evidence_result( source_job_id: JobId, ) -> FusedSearchResult | QueryAnswer: @@ -1244,7 +1388,7 @@ async def get_index_status() -> IndexStatus: description=( "Add or replace one registered media ID in the active multi-video " "index snapshot. Obtain the ID from list_media or a completed " - "upload, then poll get_job." + "upload, then observe it with wait_job and fetch get_job once." ), annotations=_SUBMIT, structured_output=True, @@ -1275,7 +1419,7 @@ def submit(actor: Principal) -> Job: title="Prepare models", description=( "Explicitly download and validate selected model artifacts. Poll " - "get_job for byte progress and completion before indexing." + "with wait_job for byte progress and completion before indexing." ), annotations=_SUBMIT, structured_output=True, @@ -1310,7 +1454,7 @@ def submit(actor: Principal) -> Job: "item in the active index snapshot. MCP returns an annotated board " "of ranked results by default. Set command.evidence_delivery.mode " "to keyframes or keyframes_and_clips only when standalone artifacts " - "are also needed, then poll only this job for results and evidence." + "are also needed, then use wait_job and get_job_evidence." ), annotations=_SUBMIT, structured_output=True, @@ -1356,12 +1500,13 @@ def submit(actor: Principal) -> Job: title="Query video", description=( "Submit a durable grounded natural-language query over indexed " - "moments and actor evidence. Set command.media_id for one video, " + "moments and actor evidence. Put the question in command.question. " + "Set command.media_id for one video, " "or omit it to query across every media item in the active index " "snapshot. MCP returns an annotated board of ranked evidence by " "default. Set command.evidence_delivery.mode to keyframes or " "keyframes_and_clips only when standalone artifacts are also needed, " - "then poll only this job for the grounded answer and evidence." + "then use wait_job and get_job_evidence." ), annotations=_SUBMIT, structured_output=True, @@ -1407,8 +1552,8 @@ def submit(actor: Principal) -> Job: title="Create clip", description=( "Create a downloadable clip from a media ID and time range returned " - "by search_moments or query_video. Poll get_job, then pass the " - "completed result's artifact_id to get_artifact_download." + "by search_moments or query_video. Wait with wait_job, then pass the " + "completed get_job result's artifact_id to get_artifact_download." ), annotations=_SUBMIT, structured_output=True, @@ -1493,10 +1638,10 @@ def submit(actor: Principal) -> Job: "Prepare keyframes and optional clips for one to ten evidence IDs " "from a completed search/query job. Use this to inspect candidates " "outside the initial bounded evidence delivery without supplying " - "timestamps or rerunning retrieval." + "timestamps or rerunning retrieval. Returns model-visible images " + "and links without duplicating the full structured result." ), annotations=_SUBMIT, - structured_output=True, ) async def materialize_job_evidence( source_job_id: JobId, @@ -1511,7 +1656,7 @@ async def materialize_job_evidence( padding_before_seconds: Annotated[float, Field(ge=0, le=30)] = 2.0, padding_after_seconds: Annotated[float, Field(ge=0, le=30)] = 2.0, profile: SnippetProfile = SnippetProfile.compatible_mp4, - ) -> Annotated[CallToolResult, EvidenceDeliveryResult]: + ) -> CallToolResult: if len(evidence_ids) != len(set(evidence_ids)): raise _application_error( ApplicationError( @@ -1545,18 +1690,18 @@ def materialize(_actor: Principal) -> EvidenceDeliveryResult: ), operation=materialize, ) - projected, blocks = await project_evidence_delivery(delivery) - if not blocks: - blocks.append( - TextContent( - type="text", - text="VidXP could not produce the requested evidence artifacts.", - ) - ) - return CallToolResult( - content=blocks, - structured_content=projected.model_dump(mode="json"), + projected_delivery, blocks = await project_evidence_delivery(delivery) + blocks.insert( + 0, + TextContent( + type="text", + text=evidence_index( + source_job_id=source_job_id, + delivery=projected_delivery, + ), + ), ) + return CallToolResult(content=blocks) @server.tool( title="Create evidence board", @@ -1635,7 +1780,8 @@ async def get_artifact_download( title="List jobs", description=( "List durable jobs and IDs so work can be recovered across sessions. " - "Use get_job for transport-projected evidence and ResourceLinks." + "Use get_job_evidence to present completed search/query evidence, or " + "get_job for the full machine record." ), annotations=_READ_ONLY, structured_output=True, @@ -1659,62 +1805,110 @@ async def list_jobs( @server.tool( title="Get job", description=( - "Poll a durable VidXP job and its typed result. Completed search " - "and query jobs include the default annotated board plus any requested " - "standalone frames or clips. Custom evidence-board jobs include their " - "annotated pages and tile map in this same response." + "Fetch the full typed machine record for a durable VidXP job. Use " + "get_job_status or wait_job while work is active. For completed " + "search/query evidence, prefer get_job_evidence so images remain " + "model-visible without duplicating this full record." ), annotations=_READ_ONLY, structured_output=True, ) - async def get_job(job_id: JobId) -> Annotated[CallToolResult, Job]: - job = await _invoke_async( + async def get_job(job_id: JobId) -> Job: + return await _invoke_async( context, default_principal=default_principal, permission=RepositoryPermission.read, operation=lambda _actor: context.jobs.get(job_id), ) - blocks: list[ImageContent | ResourceLink | TextContent] = [] - projected = job - if ( - job.state == JobState.succeeded - and job.result is not None - and job.kind in {JobKind.search, JobKind.query} - ): - result = job.result.result - delivery = result.evidence_delivery - if delivery is not None: - projected_delivery, blocks = await project_evidence_delivery(delivery) - projected = job.model_copy( - update={ - "result": job.result.model_copy( - update={ - "result": result.model_copy( - update={"evidence_delivery": projected_delivery} - ) - } - ) - } - ) - elif ( - job.state == JobState.succeeded - and job.result is not None - and job.kind == JobKind.evidence_board - ): - board, blocks = await project_evidence_board(job.result.result) - projected = job.model_copy( - update={"result": job.result.model_copy(update={"result": board})} - ) - if not blocks: - blocks.append( - TextContent( - type="text", - text=(f"VidXP job {job.job_id} is {job.state.value}."), + + @server.tool( + title="Present job evidence", + description=( + "Present a completed search, query, or evidence-board job as a " + "concise evidence index plus model-visible board images and resource " + "links. This intentionally omits structuredContent; use get_job only " + "when the full machine record is actually needed." + ), + annotations=_READ_ONLY, + ) + async def get_job_evidence(job_id: JobId) -> CallToolResult: + def completed_evidence_job(_actor: Principal) -> Job: + job = context.jobs.get(job_id) + if ( + job.state != JobState.succeeded + or job.result is None + or job.kind + not in {JobKind.search, JobKind.query, JobKind.evidence_board} + ): + raise ApplicationError( + "job_evidence_not_ready", + ErrorCategory.conflict, + "Evidence presentation requires a completed search, query, " + "or evidence-board job.", ) - ) - return CallToolResult( - content=blocks, - structured_content=projected.model_dump(mode="json"), + return job + + job = await _invoke_async( + context, + default_principal=default_principal, + permission=RepositoryPermission.read, + operation=completed_evidence_job, + ) + try: + blocks = await evidence_content(job) + except ApplicationError as exc: + raise _application_error(exc) from exc + return CallToolResult(content=blocks) + + @server.tool( + title="Get compact job status", + description=( + "Return compact durable job status without its typed result, " + "evidence payloads, or ResourceLinks. Prefer wait_job after this " + "initial observation." + ), + annotations=_READ_ONLY, + structured_output=True, + ) + async def get_job_status(job_id: JobId) -> JobSummary: + return await _invoke_async( + context, + default_principal=default_principal, + permission=RepositoryPermission.read, + operation=lambda _actor: context.jobs.summary(job_id), + ) + + @server.tool( + title="Wait for job change", + description=( + "Wait up to 30 seconds for a durable job to change stage or reach a " + "terminal state. Pass the previous observation token on subsequent " + "calls. Evidence rendering is treated as one observable phase " + "rather than waking once per artifact. Returns compact status only; " + "after completion use " + "get_job_evidence for visual search/query output, or get_job when " + "the full machine record is needed." + ), + annotations=_READ_ONLY, + structured_output=True, + ) + async def wait_job( + job_id: JobId, + after_observation_token: Sha256 | None = None, + timeout_seconds: Annotated[ + int, + Field(gt=0, le=MAX_JOB_WAIT_SECONDS), + ] = DEFAULT_JOB_WAIT_SECONDS, + ) -> JobWaitResult: + return await _invoke_async( + context, + default_principal=default_principal, + permission=RepositoryPermission.read, + operation=lambda _actor: context.jobs.wait_for_change( + job_id, + after=after_observation_token, + timeout_seconds=timeout_seconds, + ), ) @server.tool( diff --git a/tests/test_api.py b/tests/test_api.py index dbe1cea..09fd756 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -36,6 +36,7 @@ JobKind, JobQueue, JobState, + JobWaitResult, IndexStatus, MediaAsset, Principal, @@ -549,6 +550,43 @@ def test_failed_model_preparation_job_is_structured_over_http(self): self.assertTrue(error["retryable"]) self.assertTrue(error["details"]["partial_files_preserved"]) + def test_http_job_summary_and_bounded_wait_do_not_project_results(self): + with TemporaryDirectory() as directory: + context = self.context(Path(directory)) + compact = JobService._summary(queued_job()) + context.jobs.summary.return_value = compact + context.jobs.wait_for_change.return_value = JobWaitResult( + job=compact, + changed=False, + timed_out=True, + ) + with TestClient(create_app(context=context)) as client: + summary = client.get( + f"/api/v1/jobs/{JOB_ID}/status", + ) + waited = client.get( + f"/api/v1/jobs/{JOB_ID}/wait", + params={ + "after_observation_token": compact.observation_token, + "timeout_seconds": 5, + }, + ) + + self.assertEqual(summary.status_code, 200) + self.assertNotIn("result", summary.json()) + self.assertNotIn("poll_after_seconds", summary.json()) + self.assertEqual(summary.json()["observation_token"], compact.observation_token) + self.assertEqual(waited.status_code, 200) + self.assertTrue(waited.json()["timed_out"]) + self.assertNotIn("poll_after_seconds", waited.json()["job"]) + context.jobs.summary.assert_called_once_with(JOB_ID) + context.jobs.wait_for_change.assert_called_once_with( + JOB_ID, + after=compact.observation_token, + timeout_seconds=5, + ) + context.jobs.get.assert_not_called() + def test_http_job_projects_durable_evidence_to_protected_artifact_route(self): with TemporaryDirectory() as directory: context = self.context(Path(directory)) diff --git a/tests/test_job_contracts.py b/tests/test_job_contracts.py index 58720ee..2b152ef 100644 --- a/tests/test_job_contracts.py +++ b/tests/test_job_contracts.py @@ -79,6 +79,140 @@ def test_public_job_contract_has_no_path_or_storage_fields(self): self.assertNotIn('"path"', schema) self.assertNotIn("model_cache", schema) + def test_job_summary_omits_result_and_ignores_in_stage_progress_noise(self): + now = datetime.now(timezone.utc) + first = Job( + job_id=JOB_ID, + kind=JobKind.index, + state=JobState.running, + queue=JobQueue.cpu, + progress=JobProgress( + stage="frames", + message="Indexed 1 frame.", + current=1, + total=10, + updated_at=now, + ), + ) + later = first.model_copy( + update={ + "progress": first.progress.model_copy( + update={"current": 8, "message": "Indexed 8 frames."} + ) + } + ) + + first_summary = JobService._summary(first) + later_summary = JobService._summary(later) + + self.assertNotIn("result", first_summary.model_dump(mode="json")) + self.assertNotIn( + "poll_after_seconds", + first_summary.model_dump(mode="json"), + ) + self.assertFalse(first_summary.result_available) + self.assertEqual( + first_summary.observation_token, + later_summary.observation_token, + ) + self.assertEqual(later_summary.progress.current, 8) + + def test_wait_for_change_returns_on_stage_change_with_compact_status(self): + now = datetime.now(timezone.utc) + running = Job( + job_id=JOB_ID, + kind=JobKind.index, + state=JobState.running, + queue=JobQueue.cpu, + progress=JobProgress( + stage="frames", + message="Indexing frames.", + current=1, + total=10, + updated_at=now, + ), + ) + next_stage = running.model_copy( + update={ + "progress": JobProgress( + stage="embeddings", + message="Writing embeddings.", + current=0, + total=10, + updated_at=now, + ) + } + ) + backend = Mock() + backend.get.side_effect = [running, running, next_stage] + service = JobService( + settings=VidXPSettings( + repository_root=Path("repository"), + runtime_backend="cpu", + workflow_poll_interval_seconds=0.001, + ), + backend=backend, + ) + + waited = service.wait_for_change(JOB_ID, timeout_seconds=1) + + self.assertTrue(waited.changed) + self.assertFalse(waited.timed_out) + self.assertEqual(waited.job.progress.stage, "embeddings") + self.assertNotIn("result", waited.model_dump(mode="json")["job"]) + + def test_job_summary_coalesces_evidence_artifact_stages(self): + now = datetime.now(timezone.utc) + summaries = [ + JobService._summary( + Job( + job_id=JOB_ID, + kind=JobKind.search, + state=JobState.running, + queue=JobQueue.cpu, + progress=JobProgress( + stage=stage, + message=f"{stage} an evidence frame.", + updated_at=now, + ), + ) + ) + for stage in ("rendering", "validating", "publishing") + ] + + self.assertEqual( + len({summary.observation_token for summary in summaries}), + 1, + ) + self.assertEqual( + [summary.progress.stage for summary in summaries], + ["rendering", "validating", "publishing"], + ) + + def test_wait_for_change_times_out_without_fabricating_a_change(self): + running = Job( + job_id=JOB_ID, + kind=JobKind.index, + state=JobState.running, + queue=JobQueue.cpu, + ) + backend = Mock() + backend.get.return_value = running + service = JobService( + settings=VidXPSettings( + repository_root=Path("repository"), + runtime_backend="cpu", + workflow_poll_interval_seconds=0.001, + ), + backend=backend, + ) + + waited = service.wait_for_change(JOB_ID, timeout_seconds=0.003) + + self.assertFalse(waited.changed) + self.assertTrue(waited.timed_out) + self.assertEqual(waited.job.state, JobState.running) + def test_job_service_routes_model_work_without_reimplementing_it(self): backend = Mock() preflight = Mock() diff --git a/tests/test_mcp.py b/tests/test_mcp.py index 85e8cff..c6f1834 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -56,6 +56,7 @@ JobPage, JobQueue, JobState, + JobWaitResult, SearchHit, SearchJobResult, MediaPage, @@ -123,6 +124,9 @@ "get_artifact_download", "list_jobs", "get_job", + "get_job_evidence", + "get_job_status", + "wait_job", "retry_job", "cancel_job", ] @@ -422,7 +426,7 @@ def upload_context( uploads, ) - async def test_curated_tools_have_structured_output_schemas(self): + async def test_curated_tools_publish_their_intended_output_contracts(self): with TemporaryDirectory() as directory: context, _uploads = self.upload_context(Path(directory)) server = create_mcp_server( @@ -440,11 +444,26 @@ async def test_curated_tools_have_structured_output_schemas(self): [tool.name for tool in discovered.tools], MCP_TOOL_NAMES, ) + tools = {tool.name: tool for tool in discovered.tools} + self.assertIsNone(tools["get_job_evidence"].output_schema) + self.assertIsNone(tools["materialize_job_evidence"].output_schema) self.assertTrue( - all(tool.output_schema is not None for tool in discovered.tools) + all( + tool.output_schema is not None + for tool in discovered.tools + if tool.name not in {"get_job_evidence", "materialize_job_evidence"} + ) ) self.assertTrue(all(tool.title for tool in discovered.tools)) - tools = {tool.name: tool for tool in discovered.tools} + self.assertEqual( + tools["wait_job"].input_schema["properties"]["timeout_seconds"]["default"], + 30, + ) + for name in ("get_job_status", "wait_job"): + self.assertNotIn( + '"poll_after_seconds"', + json.dumps(tools[name].output_schema), + ) self.assertFalse(tools["create_media_upload"].annotations.read_only_hint) self.assertTrue(tools["create_media_upload"].annotations.idempotent_hint) self.assertTrue(tools["get_media_upload"].annotations.read_only_hint) @@ -1110,7 +1129,7 @@ def test_stdio_check_performs_handshake_and_tool_probe(self): rendered = output.getvalue() self.assertIn("OK VidXP MCP", rendered) self.assertIn("Index state: missing", rendered) - self.assertIn("Tools: 22", rendered) + self.assertIn("Tools: 25", rendered) self.assertIn("get_index_status", rendered) async def test_server_info_exposes_vidxp_branding(self): @@ -1356,6 +1375,89 @@ async def test_failed_model_preparation_job_is_structured_over_mcp(self): self.assertTrue(error["retryable"]) self.assertTrue(error["details"]["partial_files_preserved"]) + async def test_get_job_returns_machine_record_without_loading_evidence_bytes(self): + with TemporaryDirectory() as directory: + context = self.context(Path(directory)) + frame = Artifact( + artifact_id=ARTIFACT_ID, + media_id=MEDIA_ID, + generation_id="523456781234423481234567890abcde", + job_id=JOB_ID, + kind=ArtifactKind.evidence_frame, + profile="png", + mime_type="image/png", + byte_size=1, + sha256="a" * 64, + state=ArtifactState.ready, + created_at=datetime.now(timezone.utc), + ) + context.jobs.get.return_value = search_evidence_job(frame) + server = create_mcp_server( + context, + default_principal=Principal( + subject="agent", + scopes=frozenset({"vidxp.read"}), + ), + ) + async with Client(server) as client: + result = await client.call_tool("get_job", {"job_id": JOB_ID}) + + self.assertFalse(result.is_error) + self.assertEqual(result.structured_content["job_id"], JOB_ID) + self.assertFalse( + any( + isinstance(block, (ImageContent, ResourceLink)) + for block in result.content + ) + ) + context.application.open_artifact_content.assert_not_called() + + async def test_compact_job_status_and_wait_avoid_full_result_projection(self): + with TemporaryDirectory() as directory: + context = self.context(Path(directory)) + compact = JobService._summary(queued_job()) + context.jobs.summary.return_value = compact + context.jobs.wait_for_change.return_value = JobWaitResult( + job=compact, + changed=False, + timed_out=True, + ) + server = create_mcp_server( + context, + default_principal=Principal( + subject="agent", + scopes=frozenset({"vidxp.read"}), + ), + ) + async with Client(server) as client: + summary = await client.call_tool( + "get_job_status", + {"job_id": JOB_ID}, + ) + waited = await client.call_tool( + "wait_job", + { + "job_id": JOB_ID, + "after_observation_token": compact.observation_token, + "timeout_seconds": 5, + }, + ) + + self.assertFalse(summary.is_error) + self.assertNotIn("result", summary.structured_content) + self.assertNotIn("poll_after_seconds", summary.structured_content) + self.assertEqual( + summary.structured_content["observation_token"], + compact.observation_token, + ) + self.assertFalse(waited.is_error) + self.assertTrue(waited.structured_content["timed_out"]) + self.assertNotIn( + "poll_after_seconds", + waited.structured_content["job"], + ) + context.jobs.get.assert_not_called() + async def test_completed_search_projects_inline_frame_and_readable_resource(self): png_bytes = base64.b64decode( "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR4nGNg" @@ -1379,11 +1481,11 @@ async def test_completed_search_projects_inline_frame_and_readable_resource(self created_at=datetime.now(timezone.utc), ) cases = ( - ("local_stdio", False, "local_file"), - ("streamable_http", False, "mcp_resource"), - ("streamable_http", True, "https_download"), + ("local_stdio", False), + ("streamable_http", False), + ("streamable_http", True), ) - for transport, configured, expected_mode in cases: + for transport, configured in cases: with self.subTest(transport=transport, configured=configured): context = self.context( root, @@ -1419,40 +1521,35 @@ async def test_completed_search_projects_inline_frame_and_readable_resource(self artifact_delivery=transport, ) async with Client(server) as client: - result = await client.call_tool("get_job", {"job_id": JOB_ID}) + result = await client.call_tool( + "get_job_evidence", {"job_id": JOB_ID} + ) uri = f"vidxp://artifacts/{ARTIFACT_ID}/content.png" resource = await client.read_resource(uri) self.assertFalse(result.is_error) + self.assertIsNone(result.structured_content) + self.assertIn(evidence_id := "e" * 64, result.content[0].text) + self.assertIn("1.000-2.000", result.content[0].text) + if transport == "local_stdio": + self.assertIn(str(frame_path.resolve()), result.content[0].text) + elif configured: + self.assertIn( + "https://public.example/artifact-download", + result.content[0].text, + ) + else: + self.assertIn(uri, result.content[0].text) self.assertTrue( any(isinstance(block, ImageContent) for block in result.content) ) self.assertTrue( any(isinstance(block, ResourceLink) for block in result.content) ) - projected = result.structured_content["result"]["result"][ - "evidence_delivery" - ]["items"][0]["keyframe"]["artifact"] - self.assertEqual(projected["resource_uri"], uri) - delivery = projected["delivery"] - self.assertEqual(delivery["resource_uri"], uri) self.assertEqual(len(resource.contents), 1) - self.assertEqual(delivery["delivery_mode"], expected_mode) - if expected_mode == "local_file": - self.assertEqual(Path(delivery["local_path"]), frame_path) - elif expected_mode == "https_download": - self.assertTrue( - delivery["download_url"].startswith( - "https://public.example/artifact-download/" - ) - ) - self.assertIsNone(delivery["local_path"]) - else: - self.assertIsNone(delivery["local_path"]) - self.assertEqual( - delivery["delivery_error"]["code"], - "public_download_origin_unavailable", - ) + self.assertEqual( + evidence_id, durable.result.result.moments[0].moment_id + ) async def test_inline_evidence_respects_configured_resource_byte_limit(self): with TemporaryDirectory() as directory: @@ -1503,8 +1600,11 @@ async def test_inline_evidence_respects_configured_resource_byte_limit(self): artifact_delivery="streamable_http", ) async with Client(server) as client: - result = await client.call_tool("get_job", {"job_id": JOB_ID}) + result = await client.call_tool( + "get_job_evidence", {"job_id": JOB_ID} + ) self.assertFalse(result.is_error) + self.assertIsNone(result.structured_content) self.assertEqual( any( isinstance(block, ImageContent) @@ -1519,21 +1619,6 @@ async def test_inline_evidence_respects_configured_resource_byte_limit(self): ), byte_size <= maximum, ) - projected = result.structured_content["result"]["result"][ - "evidence_delivery" - ]["items"][0]["keyframe"]["artifact"] - self.assertEqual( - projected["delivery"]["delivery_mode"], - "https_download", - ) - expected_resource_uri = ( - f"vidxp://artifacts/{ARTIFACT_ID}/content.png" - if byte_size <= maximum - else None - ) - self.assertEqual( - projected["resource_uri"], expected_resource_uri - ) if not expect_inline: context.application.open_artifact_content.assert_not_called() @@ -1670,9 +1755,13 @@ async def test_materialize_job_evidence_returns_selected_frames_and_clips(self): sum(isinstance(block, ResourceLink) for block in result.content), 2, ) - item = result.structured_content["items"][0] - self.assertIsNotNone(item["keyframe"]["artifact"]["resource_uri"]) - self.assertIsNotNone(item["clip"]["resource_uri"]) + self.assertIsNone(result.structured_content) + self.assertIn("1.000-2.000", result.content[0].text) + self.assertIn("e" * 64, result.content[0].text) + self.assertIn( + f"vidxp://artifacts/{ARTIFACT_ID}/content.png", + result.content[0].text, + ) call = evidence_delivery.deliver_selected.call_args self.assertEqual(call.args[0], source.result.result) self.assertEqual(call.args[1], ("e" * 64,)) @@ -1842,22 +1931,23 @@ async def test_get_job_projects_default_evidence_board_with_search(self): ) async with Client(server) as client: - result = await client.call_tool("get_job", {"job_id": JOB_ID}) + result = await client.call_tool("get_job_evidence", {"job_id": JOB_ID}) self.assertFalse(result.is_error, result.content) + self.assertIsNone(result.structured_content) + self.assertIn("Board: 1/1 candidates", result.content[0].text) + self.assertIn("1.000-2.000", result.content[0].text) + self.assertIn("e" * 64, result.content[0].text) + self.assertIn( + f"vidxp://artifacts/{ARTIFACT_ID}/content.jpg", + result.content[0].text, + ) self.assertTrue( any(isinstance(block, ImageContent) for block in result.content) ) self.assertTrue( any(isinstance(block, ResourceLink) for block in result.content) ) - projected = result.structured_content["result"]["result"][ - "evidence_delivery" - ]["board"]["pages"][0] - self.assertEqual( - projected["artifact"]["resource_uri"], - f"vidxp://artifacts/{ARTIFACT_ID}/content.jpg", - ) async def test_one_job_search_returns_frame_and_ready_clip(self): if shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None: @@ -1928,8 +2018,10 @@ async def test_one_job_search_returns_frame_and_ready_clip(self): update={"kind": JobKind.query} ) context.jobs.get.side_effect = ( + search_evidence_job(frame, clip), search_evidence_job(frame, clip), query_evidence_job(frame, clip), + query_evidence_job(frame, clip), ) def resource(artifact_id): @@ -1973,6 +2065,9 @@ def resource(artifact_id): }, ) completed = await client.call_tool("get_job", {"job_id": JOB_ID}) + presented = await client.call_tool( + "get_job_evidence", {"job_id": JOB_ID} + ) clip_resource = await client.read_resource( f"vidxp://artifacts/{clip_id}/content.mp4" ) @@ -1991,20 +2086,29 @@ def resource(artifact_id): }, ) completed_query = await client.call_tool("get_job", {"job_id": JOB_ID}) + presented_query = await client.call_tool( + "get_job_evidence", {"job_id": JOB_ID} + ) self.assertFalse(submitted.is_error) self.assertFalse(completed.is_error) + self.assertFalse(presented.is_error) self.assertFalse(submitted_query.is_error) self.assertFalse(completed_query.is_error) + self.assertFalse(presented_query.is_error) item = completed.structured_content["result"]["result"][ "evidence_delivery" ]["items"][0] self.assertIsNotNone(item["keyframe"]) self.assertIsNotNone(item["clip"]) links = [ - block for block in completed.content if isinstance(block, ResourceLink) + block for block in presented.content if isinstance(block, ResourceLink) ] self.assertEqual(len(links), 2) + self.assertIsNone(presented.structured_content) + self.assertTrue( + any(isinstance(block, ImageContent) for block in presented.content) + ) self.assertEqual(len(clip_resource.contents), 1) duration = float( subprocess.check_output( @@ -2027,6 +2131,9 @@ def resource(artifact_id): ) query_result = completed_query.structured_content["result"]["result"] cited = query_result["claims"][0]["evidence_ids"][0] + self.assertIsNone(presented_query.structured_content) + self.assertIn("Grounded answer:", presented_query.content[0].text) + self.assertIn(cited, presented_query.content[0].text) self.assertEqual(cited, query_result["evidence"][0]["evidence_id"]) self.assertEqual( cited, @@ -2587,7 +2694,7 @@ async def test_streamable_http_works_with_the_official_remote_client(self): server.should_exit = True await serving - self.assertEqual(len(discovered.tools), 20) + self.assertEqual(len(discovered.tools), 23) self.assertNotIn( "create_media_upload", {tool.name for tool in discovered.tools},