feat: support user-defined workflow nodes

docs/examples/workflow-nodes/README.md

@@ -0,0 +1,38 @@
+# User-defined workflow nodes
+
+A workflow step can embed a template or refer to an active node published for the same Jira project. Each step has its own instance name. Project references are resolved to the active immutable node revision when the workflow starts; the resolved template and dependency digest are stored in the pinned workflow definition.
+
+For example, add this step to a feature workflow after specification generation:
+
+```yaml
+review_artifact:
+  node:
+    source: project
+    name: artifact-review
+  cases:
+    - when: {fact: node.review_artifact.outcome, op: equals, value: ready}
+      next: generate_tasks
+  otherwise: generate_spec
+```
+
+The example [artifact-review-node.yaml](artifact-review-node.yaml) selects only the ticket key and specification content. `agent-assessment-v1` makes one structured model call without agent tools or repository access. Its response must contain one declared outcome and a summary of at most 1,000 characters. Missing required input, model failure, or an invalid response blocks the run. `decision-v1` evaluates ordered workflow predicates locally and requires a declared fallback outcome.
+
+Assessment nodes use the `user_node_assessment` model-policy key, which requires
+`structured_output` and does not require tools. A failed node remains the resume
+position so `forge:retry` reruns that assessment.
+
+```sh
+forge node catalog
+forge node validate docs/examples/workflow-nodes/artifact-review-node.yaml
+forge node publish DEMO docs/examples/workflow-nodes/artifact-review-node.yaml
+forge node activate DEMO artifact-review 1
+forge node show DEMO artifact-review
+```
+
+Use `forge node history DEMO artifact-review`, `forge node list DEMO`, and `forge node rollback DEMO artifact-review 1 --expected-active-digest <digest>` to inspect and manage revisions. Activation is separate from publication. The `--expected-active-digest` argument is required when replacing an active revision.
+
+Activation and rollback validate active workflow and subworkflow consumers against
+the candidate node revision. A revision that removes an outcome used by a consumer
+is rejected before the active node changes.
+
+Outcomes are available to conditional edges as `node.<step>.outcome`. In a parallel branch the default scope reads that branch's result. A step with `join: all` can use `scope: any` or `scope: all` in its predicate to test all recorded branch outcomes. A missing branch result blocks evaluation. Template outcomes, instructions, selected inputs, and resolved revision are part of workflow change detection.

docs/examples/workflow-nodes/artifact-review-node.yaml

@@ -0,0 +1,14 @@
+apiVersion: forge/v1
+kind: Node
+metadata:
+  name: artifact-review
+  revision: 1
+  description: Classify a specification for implementation readiness.
+spec:
+  type: agent-assessment-v1
+  outcomes: [ready, revise]
+  instruction: >-
+    Assess whether the specification is clear enough to implement. Return ready
+    only when it includes concrete acceptance criteria and no unresolved blockers.
+    Summarize the reason in one short sentence.
+  inputs: [ticket.key, artifact.spec]

docs/reference/config.md

@@ -127,6 +127,7 @@ graph-node names are not accepted in Jira configuration:
 | `bug_triage` | Bug-report completeness triage |
 | `automated_review_triage` | Classification of automated review feedback |
 | `proposal_review_triage` | Classification of proposal review threads |
+| `user_node_assessment` | All authored `agent-assessment-v1` workflow nodes |
 | `task_takeover_triage` | Existing-task takeover triage |
 | `task_takeover_planning` | Existing-task implementation planning |
 | `implement_work` | Container implementation for feature, bug, and task-takeover workflows |
@@ -198,8 +199,9 @@ key before testing a workflow.
 Forge owns stage capability requirements. Every stage requires a connection
 declaring `"capabilities": ["tools"]` except the explicitly tool-free text or
 classification stages `automated_review_triage`, `proposal_review_triage`,
-`generate_pr_description`, and `sync_pr_description`. Jira policy cannot
+`generate_pr_description`, `sync_pr_description`, and `user_node_assessment`. Jira policy cannot
 weaken that requirement.
+`user_node_assessment` requires the `structured_output` capability.
 Per-target `max_output_tokens` is limited to 131072.
 
 Resolution is project stage override (`forge.model_policy`), then the separate

src/forge/cli.py

@@ -1804,6 +1804,26 @@ def main(argv: list[str] | None = None) -> int:
     workflow_delete.add_argument("name")
     workflow_delete.add_argument("--yes", action="store_true", help="Confirm deletion")
 
+    node_parser = subparsers.add_parser("node", help="Manage project workflow node templates")
+    node_subparsers = node_parser.add_subparsers(dest="node_command")
+    for action in ("validate", "publish"):
+        node_action_parser = node_subparsers.add_parser(action)
+        if action == "publish":
+            node_action_parser.add_argument("project_key")
+        node_action_parser.add_argument("file")
+    for action in ("activate", "rollback"):
+        node_action_parser = node_subparsers.add_parser(action)
+        node_action_parser.add_argument("project_key")
+        node_action_parser.add_argument("name")
+        node_action_parser.add_argument("revision", type=int)
+        node_action_parser.add_argument("--expected-active-digest")
+    for action in ("show", "history"):
+        node_action_parser = node_subparsers.add_parser(action)
+        node_action_parser.add_argument("project_key")
+        node_action_parser.add_argument("name")
+    node_subparsers.add_parser("catalog")
+    node_subparsers.add_parser("list").add_argument("project_key")
+
     # project-setup command
     setup_parser = subparsers.add_parser(
         "project-setup",
@@ -2046,6 +2066,11 @@ def main(argv: list[str] | None = None) -> int:
         skills_parser.print_help()
         return 0
 
+    if args.command == "node":
+        from forge.workflow.declarative.node_cli import cmd_node
+
+        return asyncio.run(cmd_node(args))
+
     if args.command == "workflow":
         if getattr(args, "workflow_command", None) is None:
             workflow_parser.print_help()

src/forge/models/model_policy.py

@@ -90,6 +90,7 @@ def trace_metadata(self) -> dict[str, Any]:
     "task_takeover_review",
     "task_takeover_triage",
     "update_docs",
+    "user_node_assessment",
 )
 
 # Requirements belong to Forge's execution stages, not Jira project policy.
@@ -99,6 +100,7 @@ def trace_metadata(self) -> dict[str, Any]:
     "generate_pr_description",
     "proposal_review_triage",
     "sync_pr_description",
+    "user_node_assessment",
 }
 REQUIRED_CAPABILITIES_BY_POLICY_KEY: dict[str, frozenset[str]] = {
     key: frozenset({"tools"})
@@ -112,6 +114,7 @@ def trace_metadata(self) -> dict[str, Any]:
     "generate_tasks",
     "proposal_review_triage",
     "task_takeover_triage",
+    "user_node_assessment",
 }
 for _key in _STRUCTURED_OUTPUT_POLICY_KEYS:
     REQUIRED_CAPABILITIES_BY_POLICY_KEY[_key] = REQUIRED_CAPABILITIES_BY_POLICY_KEY.get(

src/forge/workflow/base.py

@@ -94,6 +94,17 @@ class PublicationRef(TypedDict, total=False):
     status: str
 
 
+def merge_node_results(
+    left: dict[str, dict[str, Any]] | None, right: dict[str, dict[str, Any]] | None
+) -> dict[str, dict[str, Any]]:
+    """Keep the latest attempt for each node and branch during fan-in."""
+    result = dict(left or {})
+    for key, value in (right or {}).items():
+        if key not in result or int(value.get("attempt", 0)) >= int(result[key].get("attempt", 0)):
+            result[key] = value
+    return result
+
+
 class BaseState(TypedDict, total=False):
     """State shared by ALL workflows."""
 
@@ -168,6 +179,7 @@ class BaseState(TypedDict, total=False):
     validations: list[ValidationResult]
     publications: list[PublicationRef]
     node_outcome: str | None
+    node_results: Annotated[dict[str, dict[str, Any]], merge_node_results]
 
 
 class HandoffState(TypedDict):

src/forge/workflow/declarative/cli.py

@@ -17,14 +17,18 @@
     render_mermaid,
     simulate_process_migration,
 )
+from forge.workflow.declarative.node_publication import NodePublisher
 from forge.workflow.declarative.publication import DefinitionPublisher
 
 
 async def _load_resolved(path: str, project_key: str | None, state_profile: str | None = None):
     source = load_workflow_file(path)
     publisher = DefinitionPublisher(project_key) if project_key else None
     return await resolve_definition(
-        source, publisher.active if publisher else None, state_profile=state_profile
+        source,
+        publisher.active if publisher else None,
+        state_profile=state_profile,
+        node_lookup=NodePublisher(project_key).active if project_key else None,
     )
 
 
@@ -231,10 +235,20 @@ async def cmd_workflow(args: Any) -> int:
             if active_definition is None:
                 raise ValueError(f"workflow '{args.name}' is not defined for {project_key}")
             if active_definition.kind == "Workflow":
-                resolved = await resolve_definition(active_definition, publisher.active)
+                resolved = await resolve_definition(
+                    active_definition,
+                    publisher.active,
+                    node_lookup=NodePublisher(project_key).active,
+                )
                 DeclarativeWorkflowCompiler(resolved).validate()
             else:
-                validate_subworkflow(await resolve_definition(active_definition, publisher.active))
+                validate_subworkflow(
+                    await resolve_definition(
+                        active_definition,
+                        publisher.active,
+                        node_lookup=NodePublisher(project_key).active,
+                    )
+                )
             if getattr(args, "json", False):
                 print(json.dumps(active_definition.canonical_dict(), indent=2))
             else:

src/forge/workflow/declarative/compiler.py

@@ -15,7 +15,7 @@
     effect_capability_scope,
 )
 from forge.workflow.declarative.catalog import get_state_profile
-from forge.workflow.declarative.models import MAX_TRANSITIONS, WorkflowDefinition
+from forge.workflow.declarative.models import MAX_TRANSITIONS, NodeTemplate, WorkflowDefinition
 from forge.workflow.declarative.predicates import evaluate_predicate, validate_predicate
 from forge.workflow.declarative.router_contracts import validate_router_outcomes
 from forge.workflow.preconditions import NodeContract, project_capabilities, with_preconditions
@@ -30,6 +30,17 @@ def __init__(self, definition: WorkflowDefinition) -> None:
         self.definition = definition
         self.profile = get_state_profile(definition.spec.state)
 
+    @staticmethod
+    def _has_parallel_scope(predicate: Any) -> bool:
+        return predicate.scope != "current" or any(
+            DeclarativeWorkflowCompiler._has_parallel_scope(child)
+            for child in (
+                *predicate.all_of,
+                *predicate.any_of,
+                *((predicate.not_,) if predicate.not_ else ()),
+            )
+        )
+
     def dynamic_targets(self, step: Any) -> frozenset[str]:
         """Return catalog-owned targets, accepting matching legacy metadata."""
         if not step.dynamic_route or not step.route:
@@ -85,12 +96,20 @@ def validate(self) -> None:
                 f"unsupported extension point '{sorted(unknown_extensions)[0]}'"
             )
 
-        unknown_nodes = set(steps) - set(self.profile.nodes)
+        unknown_nodes = {
+            name
+            for name, step in steps.items()
+            if name not in self.profile.nodes and not isinstance(step.node, NodeTemplate)
+        }
         if unknown_nodes:
             raise WorkflowValidationError(
                 f"node '{sorted(unknown_nodes)[0]}' is not registered for state '{spec.state}'"
             )
-        missing_effect_policies = set(steps) - set(self.profile.effect_policies)
+        missing_effect_policies = {
+            name
+            for name, step in steps.items()
+            if name not in self.profile.effect_policies and not isinstance(step.node, NodeTemplate)
+        }
         if missing_effect_policies:
             raise WorkflowValidationError(
                 f"node '{sorted(missing_effect_policies)[0]}' has no registered effect policy"
@@ -99,6 +118,16 @@ def validate(self) -> None:
         adjacency: dict[str, set[str]] = {name: set() for name in steps}
         has_terminal = False
         for node_name, step in steps.items():
+            if step.node is not None:
+                if node_name in self.profile.nodes or not isinstance(step.node, NodeTemplate):
+                    raise WorkflowValidationError(
+                        f"node template for '{node_name}' is unresolved or shadows a trusted node"
+                    )
+                if step.route or step.dynamic_route:
+                    raise WorkflowValidationError("user node cannot use trusted routers")
+                if step.node.type == "decision-v1":
+                    for rule in step.node.rules:
+                        validate_predicate(rule.when, spec.state, steps=steps)
             if step.route and step.route not in self.profile.routers:
                 raise WorkflowValidationError(
                     f"router '{step.route}' on '{node_name}' is not registered for state "
@@ -119,8 +148,10 @@ def validate(self) -> None:
                         f"protected gate '{node_name}' requires a trusted router"
                     )
                 for case in step.cases:
+                    if self._has_parallel_scope(case.when) and step.join != "all":
+                        raise WorkflowValidationError("cross-branch outcome requires join: all")
                     try:
-                        validate_predicate(case.when, spec.state)
+                        validate_predicate(case.when, spec.state, steps=steps)
                     except ValueError as exc:
                         raise WorkflowValidationError(f"step '{node_name}' {exc}") from exc
             for target in targets:
@@ -161,7 +192,9 @@ def validate(self) -> None:
                     f"node '{node_name}' does not support station contract "
                     f"'{step.station_contract}'"
                 )
-            if step.kind is not None and step.kind != self.profile.node_kind(node_name):
+            if step.kind is not None and step.kind != (
+                "operation" if step.node else self.profile.node_kind(node_name)
+            ):
                 raise WorkflowValidationError(
                     f"node kind for '{node_name}' is catalog-owned and must be "
                     f"'{self.profile.node_kind(node_name)}'"
@@ -260,7 +293,7 @@ def build_graph(self) -> StateGraph[Any]:
             graph.add_node(
                 node_name,
                 self._guarded_node(
-                    self.profile.nodes[node_name],
+                    self._node_function(node_name),
                     node_name,
                     terminal=step.next == "__end__",
                     contract=self.profile.contracts.get(node_name),
@@ -324,8 +357,31 @@ def build_graph(self) -> StateGraph[Any]:
     def effective_effects(self, node_name: str) -> tuple[str, ...]:
         """Resolve catalog-owned authority and an optional supported restriction."""
         step = self.definition.spec.steps[node_name]
+        if isinstance(step.node, NodeTemplate):
+            if step.allowed_effects:
+                raise ValueError("user nodes cannot request effects")
+            return ()
         return self.profile.effect_policies[node_name].resolve(step.allowed_effects)
 
+    def _node_function(self, node_name: str) -> Callable[..., Any]:
+        step = self.definition.spec.steps[node_name]
+        if isinstance(step.node, NodeTemplate):
+            from forge.workflow.declarative.user_nodes import execute_node
+
+            async def run(state: dict[str, Any]) -> dict[str, Any]:
+                try:
+                    return await execute_node(step.node, state, node_name)
+                except Exception as exc:
+                    return {
+                        **state,
+                        "current_node": node_name,
+                        "is_blocked": True,
+                        "last_error": f"User node {node_name}: {exc}",
+                    }
+
+            return run
+        return self.profile.nodes[node_name]
+
     def effective_observation_policy(self) -> str | None:
         """Return the profile policy implied by this definition's topology."""
         return self.profile.observation_policy_for(set(self.definition.spec.steps))