Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,7 @@ one to open.
| Worker protocol | `dw/worker_protocol.py`: `parse_reply` | Every command and reply is a frozen dataclass that travels as a wire dict. | `tests/test_worker_messages.py::test_from_wire_inverts_to_wire`, `tests/test_worker_messages.py::test_an_unknown_reply_type_is_kept_whole_rather_than_raised` |
| Persistent worker | `dw/worker.py`, `dw/worker_manager.py`, `dw/serve.py` | Jobs run in one spawned worker that keeps models loaded, so a change to engine code needs a server restart. | — |
| Failed-run reporting | `dw/worker.py`, `dw/worker_protocol.py`: `Failed`, `Cancelled` | A failed or cancelled run's reply still carries the manifest of the steps that ran. | `tests/test_worker_execute.py::test_failure_carries_the_manifest_of_the_steps_that_ran`, `tests/test_worker_execute.py::test_cancellation_carries_the_manifest_too` |
| Failure path redaction | `dw/path_redaction.py`: `redact_paths`, called by `dw/worker.py` | A failed run's message and traceback name a file under the job's asset roots or output directory by its `asset:`/`output:` reference, never its absolute server path. | `tests/test_worker_execute.py::test_a_failure_names_an_asset_by_reference_not_by_server_path`, `tests/test_path_redaction.py` |
| Run-time warnings | `dw/events.py`: `emit_warning` | A warning found at run time is emitted as an event, so it reaches the caller and not just the log. | `tests/test_concat_videos.py::TestWarningsReachTheCaller::test_the_level_spread_warning_is_emitted_as_an_event` |

## Media and DSP
Expand Down
2 changes: 1 addition & 1 deletion docs/MCP.md
Original file line number Diff line number Diff line change
Expand Up @@ -272,7 +272,7 @@ tools) rather than relying on the session pin.
| `validate_workflow(workflow=None, name=None, workspace=None, arguments=None)` | exactly one of `workflow` (inline definition) or `name` (a stored workflow, as `list_workflows` reports it), optional `workspace`, optional `arguments` | Check a workflow against the schema and against real pipeline signatures. Free and instant. Validating by name uses the workflow file's own directory as the base directory, so it sees what a run would. Returns every schema violation in `errors`, each with the JSON path it sits at, so a draft is fixed in one pass, and a `previous_result:` that names no earlier step is one of them. `warnings` covers what still runs but is probably wrong - a signature mismatch, and, for a list-driven variable, an entry key no step reads, at the entry's path. `workspace` names the workspace for this one call without switching the session to it - use it to pin a job whose `output:` or `asset:` references live in a workspace other than the session's. Pass the same `arguments` you will pass to `run_workflow` and they are checked too - an undeclared or renamed variable name, a value that will not coerce to the declared type, and an `asset:`, `prompt:` or `output:` reference that names nothing this workspace can reach, each reported at `arguments.<name>`. `checked_arguments` lists what was covered, so a `valid: true` about the stored defaults cannot be mistaken for one about your values. A reference set the model would refuse - too many images, videos or audio clips, or, for MiniMax-H3, audio as the only reference - is an error here too, rather than a failure minutes into a run you acknowledged. `run_workflow` makes the same check and refuses a bad argument rather than queuing a job that fails on its first step. A valid answer carries `plan` - the fingerprint, step count, list lengths, `downloads_required`, `estimate` (with `basis`) and `elided_steps` for the arguments given; quote from it. `steps` counts what will run: a step nothing reads and which saves no file does not run, and is named in `elided_steps` instead |
| `list_workspaces()` | — | The server's workspaces and which one this session is using. Each has its own workflows, assets and outputs; the prompt library is shared by all of them |
| `use_workspace(name)` | `name` | Work in that workspace for the rest of the session - every later call reads and writes there. This is how to keep your work out of another agent's namespace rather than sharing the default one. Checked against the server, so a typo fails here rather than scoping every later call to nothing. On a `dw.serve --mcp` endpoint the pin is shared by every connected client (#298, see the note above the table) - a `warning` field appears when this call overwrote a pin already pointed away from `default` |
| `create_workspace(name, use=False)` | `name`, `use` | Create a workspace. Pass use=true to switch this session to it as well; otherwise the session stays where it was and the result says so. `use=true` carries the same mounted-server caveat and `warning` field as `use_workspace` |
| `create_workspace(name, use=False)` | `name`, `use` | Create a workspace. Pass use=true to switch this session to it as well; otherwise the session stays where it was and the result says so. The result names the workspace (`name`, `default`, `current`, `next`), not its folders - `list_workspaces(detail=true)` is the opt-in for those. `use=true` carries the same mounted-server caveat and `warning` field as `use_workspace` |
| `delete_workspace(name, acknowledged_cost=False)` | `name`, `acknowledged_cost` | Permanently delete a workspace and everything in it. Refuses without the acknowledgement, reporting what it would remove |
| `list_assets()` | — | The input media on the server, each with the `asset:` reference a workflow argument carries. Look here before asking for a file - what a workflow needs may already be there. `libraries` names the roots searched and which are writable; `shadowed` lists names a nearer library hides |
| `keep_output(name, asset_name=None, overwrite=False, shared=False, workspace=None)` | `name`, optional `asset_name`, `overwrite`, `shared`, `workspace` | Keep a generated file as an input asset under a stable `asset:` name, so a later workflow can rely on it. The copy happens on the server: nothing is downloaded or re-uploaded. `asset_name` may name a folder and takes the kept file's extension when it has none; `shared=true` keeps it in the library every workspace shares, which is where a recurring cast belongs. `workspace` names the workspace for this one call without switching the session to it - the same pin `run_workflow` takes, so a job run into another workspace stays reachable from the session that queued it |
Expand Down
2 changes: 1 addition & 1 deletion docs/stabilization/baseline.json
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
{
"modules": 168,
"modules": 171,
"modules_over_size_ceiling": 0,
"functions_over_150_lines": 0,
"prefix_literals": 0,
Expand Down
7 changes: 5 additions & 2 deletions dw/assets.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,9 +121,12 @@ def resolve_asset_reference(reference, asset_dir=None, base_dir=None, library=No
if found:
logger.debug(f"Resolved {reference} to {found[0]}")
return found[0]
searched = ", ".join(root.root for root in library.roots())
# Named by origin, not directory: this message reaches API and MCP
# callers, and the roots are the server's filesystem layout
searched = ", ".join(dict.fromkeys(root.origin for root in library.roots()))
raise ValueError(
f"Asset '{name}' not found in {searched} - an 'asset:' reference "
f"Asset '{name}' not found in the asset library ({searched}) - an "
f"'asset:' reference "
f"names a file in the asset library, with its extension, like "
f"'asset:iris.jpg' or 'asset:gyre/frame_1.jpg'"
)
Expand Down
6 changes: 6 additions & 0 deletions dw/events.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,12 @@ def note_phase(self, phase):
by the watchdog to time how long the current phase has run."""
self._current_phase = phase
self._phase_started_at = time.monotonic()
# A new phase starts its own silence: measured from the last event
# instead, a quiet stretch before the phase made the watchdog report
# a stall at +0.0s into it - for a check landing before emit_phase's
# own event, or on every check for a caller that notes a phase
# without emitting one
self._last_event_at = self._phase_started_at

def enter_run(self):
"""Called around Workflow.run - starts the watchdog on the
Expand Down
72 changes: 72 additions & 0 deletions dw/path_redaction.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
"""A failed run's message, with the server's library paths named back as
references.

A task that fails on an input reports the file it was handed, and by then an
'asset:' reference has been resolved to an absolute path - so a job error
read over MCP or the HTTP API carried the server's home directory and
workspace layout (GHSA-fwg5-jfjg-fxpf). The message is rewritten where the
worker builds the failure reply, once, rather than in every task that might
name a path: a file under an asset root becomes 'asset:<name>', one under
the job's output directory 'output:<name>' - the reference the caller would
write to address it - and a bare root becomes the library's name.

It imports only dw.references - which imports nothing from dw - so the
worker's failure path cannot fail for want of it, and the reference is built
by the module that owns its spelling.
"""

import os
import re

from . import references

# What follows a root in a message, up to the end of the path: anything but
# whitespace or the quotes a message wraps a path in. A name with a space in
# it is cut at the space, so the reference reads short - the root is gone
# either way
_REST_OF_PATH = r"([^\s'\"`]*)"


def _spellings(root):
"""The ways a root may appear in a message: as configured, and resolved
(a path validate_path returned has had its symlinks followed - /tmp is
/private/tmp on macOS)."""
spellings = {os.path.abspath(root), os.path.realpath(root)}
return {spelling.rstrip(os.sep) for spelling in spellings if spelling}


def redact_paths(text, asset_roots=(), output_dir=None):
"""`text` with every path under the given roots written as a reference.

Args:
text: A message or traceback; None or "" is returned as it came
asset_roots: The directories on the job's asset search path
output_dir: The job's output directory
"""
if not text:
return text
rewrites = []
for root in asset_roots or ():
if root:
rewrites += [
(s, references.ASSET, "the asset library") for s in _spellings(root)
]
if output_dir:
rewrites += [
(s, references.OUTPUT, "the output folder") for s in _spellings(output_dir)
]
# Longest first, so a root nested in another is rewritten by its own
# prefix rather than as a subfolder of the outer one
for root, kind, label in sorted(rewrites, key=lambda r: -len(r[0])):
text = re.sub(
re.escape(root + os.sep) + _REST_OF_PATH,
# A reference's separator is '/', whatever the server's is
lambda match, kind=kind: references.make_ref(
kind, match.group(1).replace(os.sep, "/")
),
text,
)
# Only the root itself: '/ws/assets' must not turn '/ws/assets2'
# into 'the asset library2'
text = re.sub(re.escape(root) + r"(?![\w-]|\.\w)", label, text)
return text
4 changes: 2 additions & 2 deletions dw/server/admission.py
Original file line number Diff line number Diff line change
Expand Up @@ -272,8 +272,8 @@ def _string_leaves(value, path):
"this workspace has no asset library"
)
# One resolve over the whole path, the way a run does:
# a miss names every root it looked in, the workspace's
# own first
# a miss names every library it looked in, the
# workspace's own first
resolve_asset_reference(leaf, library=asset_library)
elif references.is_ref(references.PROMPT, leaf):
resolve_prompt_reference(leaf, library=prompt_library)
Expand Down
11 changes: 8 additions & 3 deletions dw/server/outputs.py
Original file line number Diff line number Diff line change
Expand Up @@ -247,11 +247,16 @@ def asset_in(name, library):
found = library.find(name)
if found:
return found[0]
roots = [root.root for root in library.roots()]
if not roots:
# Named by origin, not directory: the roots are the server's filesystem
# layout (GHSA-fwg5-jfjg-fxpf)
origins = list(dict.fromkeys(root.origin for root in library.roots()))
if not origins:
detail = f"Unknown asset {name!r}: this workspace has no asset library"
else:
detail = f"Unknown asset {name!r}: not found in {', '.join(roots)}"
detail = (
f"Unknown asset {name!r}: not found in the asset library "
f"({', '.join(origins)})"
)
raise HTTPException(status_code=404, detail=detail)


Expand Down
6 changes: 3 additions & 3 deletions dw/server/routes/files.py
Original file line number Diff line number Diff line change
Expand Up @@ -71,9 +71,9 @@ async def output_file(
name = strip_output_prefix(name)
if is_asset_reference(name):
# This route is outside the token gate (the auth middleware
# covers /api/ and /mcp only), so a miss must not carry
# _asset_file's detail, which names every root searched by its
# absolute server path. Keep the hint #445 added, without them.
# covers /api/ and /mcp only), so a miss says no more than that
# the name is not in this workspace's library - not even which
# libraries were searched. Keep the hint #445 added.
try:
path = asset_file(request.app.state, name, ws)
except HTTPException as e:
Expand Down
27 changes: 24 additions & 3 deletions dw/worker.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,8 @@
from dw import workflow_run
from dw.workflow import workflow_from_snapshot
from dw.step_cache import step_cache
from dw.assets import activate_asset_dir, deactivate_asset_dir
from dw.assets import activate_asset_dir, asset_library, deactivate_asset_dir
from dw.path_redaction import redact_paths
from dw.log_setup import setup_logging, set_log_level
from dw.settings import load_settings, resolve_path
from dw.events import RunContext, WorkflowCancelled
Expand Down Expand Up @@ -282,9 +283,12 @@ def _handle_execute(self, command: Dict[str, Any]):
)
except Exception as e:
logger.error(f"Error executing workflow: {e}", exc_info=True)
# The log keeps the paths; the reply reaches API and MCP
# callers, who address files by reference (GHSA-fwg5-jfjg-fxpf)
redact = self._failure_redactor(command)
failure = Failed(
message=f"Workflow execution error: {str(e)}",
traceback=traceback.format_exc(),
message=redact(f"Workflow execution error: {str(e)}"),
traceback=redact(traceback.format_exc()),
# The files the steps before the failure wrote are on
# disk; reporting them is what keeps a run that died at
# step five from looking like one that produced nothing
Expand All @@ -307,6 +311,23 @@ def _handle_execute(self, command: Dict[str, Any]):
if job.asset_token is not None:
deactivate_asset_dir(job.asset_token)

@staticmethod
def _failure_redactor(command: Dict[str, Any]):
"""What rewrites a failure reply's paths under this job's asset
search path and output directory as references - called while the
job's asset root is still active, so the search path is its own.
A search path that cannot be built leaves the asset roots out
rather than failing the failure report."""
try:
asset_roots = [
root.root for root in asset_library(command.get("asset_dir")).roots()
]
except Exception:
logger.debug("No asset search path to redact against", exc_info=True)
asset_roots = [command["asset_dir"]] if command.get("asset_dir") else []
output_dir = command.get("output_dir")
return lambda text: redact_paths(text, asset_roots, output_dir)

def _activate_job(self, command: Dict[str, Any], job: "_Job"):
"""Phase 1: the job's log level and its asset root, before anything
that could resolve an 'asset:' reference (B8)."""
Expand Down
14 changes: 12 additions & 2 deletions dw_mcp/workspaces.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,11 @@
# size, nothing that only naming its folders needs
WORKSPACE_SUMMARY_FIELDS = ("name", "default", "usage")

# What create_workspace answers with: the name to address it by, never the
# folders the server made for it - no MCP response carries an absolute
# server path unasked (#521, #527, GHSA-crqf-hw9p-r739)
CREATED_WORKSPACE_FIELDS = ("name", "default")

# dw.serve --mcp builds one DwClient for every connected agent (#298) - the
# pin is server-global there, not per-session, and that is a deliberate scope
# decision (this server is single-user) rather than a bug to fix with session
Expand Down Expand Up @@ -122,8 +127,13 @@ def create_workspace(client, name, use=False):
and outputs, and shares the server's one prompt library. Creating it
does not switch to it unless `use` is true - the natural
create-then-run sequence otherwise runs in the workspace the session
was already in, and the result says which that is."""
body = client.post_json("/api/workspaces", {"name": name})
was already in, and the result says which that is.

The answer names the workspace, not its folders; list_workspaces
(detail=True) is the opt-in for those."""
body = project(
client.post_json("/api/workspaces", {"name": name}), CREATED_WORKSPACE_FIELDS
)
warning = _switch(client, name) if use else None
result = {
**body,
Expand Down
19 changes: 19 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,25 @@ def _isolate_settings_dir(tmp_path, monkeypatch):
monkeypatch.setenv("DIFFUSERS_HELPER_ROOT", str(tmp_path / "helper"))


@pytest.fixture(autouse=True)
def _isolate_device_capacity(monkeypatch):
"""Check a vram_estimate against the catalog's measured cards, never the
accelerator the suite happens to run on.

The check uses the serving device's own capacity when no 'cost' entry
describes its backend (dw/vram_estimate.py `_entries_for`), so on a
24 GB Mac - a 17.8 GB Metal working set - a template's own defaults were
refused and its tests failed there and nowhere else. With no capacity
read, every machine gets what CI gets. A test about the device ceiling
patches its own capacity, which wins over this one.
"""
import dw.validation
import dw.workflow_run

for module in (dw.validation, dw.workflow_run):
monkeypatch.setattr(module, "device_capacity_gb", lambda device=None: None)


@pytest.fixture(autouse=True)
def _clear_task_model_cache():
"""Ensure dw.tasks.model_cache is empty at the start of every test.
Expand Down
22 changes: 22 additions & 0 deletions tests/test_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -586,3 +586,25 @@ def test_select_kinds_matches_a_warnings_kind_as_well_as_event():
assert [e["seq"] for e in select_kinds(events, ["log", "warning"])] == [1, 2, 3]
assert [e["seq"] for e in select_kinds(events, None)] == [0, 1, 2, 3]
assert select_kinds(events, ["nothing"]) == []


def test_a_quiet_stretch_before_a_phase_is_not_that_phases_stall():
"""A watchdog that measured silence from the last event, not from the
phase start, reported a stall at +0.0s into a phase noted after a quiet
stretch - how test_watchdog_fires_after_threshold_with_no_events failed
under a loaded parallel run."""
events = []
context = RunContext(on_event=events.append)
with _fast_watchdog():
context.enter_run()
try:
# longer than the threshold, before any phase
time.sleep(0.1)
context.note_phase("generating")
time.sleep(0.2)
finally:
context.exit_run()

stalls = _stalls(events)
assert stalls
assert stalls[0]["seconds_since_phase_start"] > 0
6 changes: 5 additions & 1 deletion tests/test_library_sources.py
Original file line number Diff line number Diff line change
Expand Up @@ -264,7 +264,11 @@ def test_validated_arguments_reach_the_whole_asset_path(self, client, tmp_path):

assert missing["valid"] is False
assert missing["errors"][0]["path"] == "arguments.image"
assert os.path.abspath(workspace.assets) in missing["errors"][0]["message"]
# The miss names the libraries it searched, never their server
# directories (GHSA-fwg5-jfjg-fxpf)
message = missing["errors"][0]["message"]
assert "asset library (workspace, examples)" in message
assert os.path.abspath(workspace.assets) not in message


def test_two_apps_in_one_process_each_list_their_own_examples(tmp_path):
Expand Down
Loading
Loading