perago.invoke_workspace_task_body

perago.invoke_workspace_task_body(task, input_data, workspace_dir)

Invoke a workspace task body against a prepared local workspace.

The helper performs the task-body portion of a workspace attempt: it checks the Conductor input wrapper, validates WorkspaceInput and params, applies pre guardrails, calls the task function, validates the declared output model, and applies post guardrails.

Parameters:

taskTaskDefinition

Loaded workspace task definition.

input_datamapping of str to Any

Conductor task input containing exactly "workspace" and "params".

workspace_dirpathlib.Path

Attempt-local workspace directory already populated from the workspace input.

Returns:

dict of str to Any

Body output wrapper containing only the validated "result" field.

Raises:

TaskInputError

If task is not a workspace task or if the input wrapper shape is invalid.

PreGuardrailViolation

If a pre-execution workspace guardrail fails.

PostGuardrailViolation

If a post-execution workspace guardrail fails.

pydantic.ValidationError

If workspace input, params, or task result validation fails.

参数:

• task (TaskDefinition)

• input_data (Mapping[str, Any])

• workspace_dir (Path)

返回类型:

dict[str, Any]

参见

run_workspace_task_attempt

Full workspace attempt execution flow.

build_workspace_task_output

Add a published workspace ref to a validated result.

check_guardrails

Evaluate workspace guardrail declarations.

Examples

>>> from pathlib import Path
>>> task_def = load_module_task("app.workers.features_build")
>>> output = invoke_workspace_task_body(
...     task_def,
...     {"workspace": {...}, "params": {"feature_set": "default", "min_rows": 100}},
...     Path("/tmp/perago/workspaces/attempt"),
... )
>>> sorted(output)
['result']