From fa06db1dfa064cb081bbd2b5e8d3bfc61ea8a76d Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Wed, 30 Sep 2026 16:33:45 +0300 Subject: [PATCH 1/2] feat: support project-authored workflow node templates --- docs/examples/workflow-nodes/README.md | 30 +++ .../workflow-nodes/artifact-review-node.yaml | 14 ++ src/forge/cli.py | 25 ++ src/forge/workflow/base.py | 12 + src/forge/workflow/declarative/cli.py | 20 +- src/forge/workflow/declarative/compiler.py | 67 ++++- src/forge/workflow/declarative/composition.py | 43 +++- src/forge/workflow/declarative/manifest.py | 3 +- src/forge/workflow/declarative/models.py | 88 ++++++- src/forge/workflow/declarative/node_cli.py | 81 ++++++ .../workflow/declarative/node_publication.py | 186 ++++++++++++++ src/forge/workflow/declarative/predicates.py | 58 ++++- src/forge/workflow/declarative/publication.py | 61 ++++- src/forge/workflow/declarative/resolver.py | 27 +- src/forge/workflow/declarative/user_nodes.py | 75 ++++++ tests/unit/workflow/test_user_nodes.py | 234 ++++++++++++++++++ 16 files changed, 990 insertions(+), 34 deletions(-) create mode 100644 docs/examples/workflow-nodes/README.md create mode 100644 docs/examples/workflow-nodes/artifact-review-node.yaml create mode 100644 src/forge/workflow/declarative/node_cli.py create mode 100644 src/forge/workflow/declarative/node_publication.py create mode 100644 src/forge/workflow/declarative/user_nodes.py create mode 100644 tests/unit/workflow/test_user_nodes.py diff --git a/docs/examples/workflow-nodes/README.md b/docs/examples/workflow-nodes/README.md new file mode 100644 index 00000000..f0cf38cc --- /dev/null +++ b/docs/examples/workflow-nodes/README.md @@ -0,0 +1,30 @@ +# 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. + +```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 ` to inspect and manage revisions. Activation is separate from publication. The `--expected-active-digest` argument is required when replacing an active revision. + +Outcomes are available to conditional edges as `node..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. diff --git a/docs/examples/workflow-nodes/artifact-review-node.yaml b/docs/examples/workflow-nodes/artifact-review-node.yaml new file mode 100644 index 00000000..1ff0d9b4 --- /dev/null +++ b/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] diff --git a/src/forge/cli.py b/src/forge/cli.py index 2ff507d2..2f501938 100644 --- a/src/forge/cli.py +++ b/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() diff --git a/src/forge/workflow/base.py b/src/forge/workflow/base.py index 215b4ea4..9cc68578 100644 --- a/src/forge/workflow/base.py +++ b/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): diff --git a/src/forge/workflow/declarative/cli.py b/src/forge/workflow/declarative/cli.py index 91cd8a2b..fd7826ae 100644 --- a/src/forge/workflow/declarative/cli.py +++ b/src/forge/workflow/declarative/cli.py @@ -17,6 +17,7 @@ render_mermaid, simulate_process_migration, ) +from forge.workflow.declarative.node_publication import NodePublisher from forge.workflow.declarative.publication import DefinitionPublisher @@ -24,7 +25,10 @@ async def _load_resolved(path: str, project_key: str | None, state_profile: str 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: diff --git a/src/forge/workflow/declarative/compiler.py b/src/forge/workflow/declarative/compiler.py index a514d2fe..2171548f 100644 --- a/src/forge/workflow/declarative/compiler.py +++ b/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,30 @@ 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, + "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)) diff --git a/src/forge/workflow/declarative/composition.py b/src/forge/workflow/declarative/composition.py index e6f06180..e2ab5ab6 100644 --- a/src/forge/workflow/declarative/composition.py +++ b/src/forge/workflow/declarative/composition.py @@ -4,14 +4,23 @@ from collections.abc import Awaitable, Callable -from forge.workflow.declarative.models import MAX_STEPS, WorkflowDefinition, WorkflowStep +from forge.workflow.declarative.models import ( + MAX_STEPS, + NodeDefinition, + NodeReference, + NodeTemplate, + WorkflowDefinition, + WorkflowStep, +) ProjectLookup = Callable[[str], Awaitable[WorkflowDefinition | None]] +NodeLookup = Callable[[str], Awaitable[NodeDefinition | None]] def validate_subworkflow(definition: WorkflowDefinition) -> None: """Check fragment authority without requiring a standalone terminal graph.""" from forge.workflow.declarative.catalog import get_state_profile + from forge.workflow.declarative.predicates import validate_predicate if definition.kind != "Subworkflow": raise ValueError("expected a subworkflow") @@ -24,7 +33,7 @@ def validate_subworkflow(definition: WorkflowDefinition) -> None: for state in definition.spec.compatible_states or (definition.spec.state,): profile = get_state_profile(state) for name, step in definition.spec.steps.items(): - if name not in profile.nodes: + if name not in profile.nodes and not isinstance(step.node, NodeTemplate): raise ValueError(f"node '{name}' is not registered for state '{state}'") if step.route and step.route not in profile.routers: raise ValueError(f"router '{step.route}' is not registered for state '{state}'") @@ -33,13 +42,17 @@ def validate_subworkflow(definition: WorkflowDefinition) -> None: validate_router_outcomes(state, name, step.route, set(step.branches)) if step.cases: - from forge.workflow.declarative.predicates import validate_predicate - if name in profile.mandatory_nodes or name.endswith("_gate"): raise ValueError(f"protected gate '{name}' requires a trusted router") for case in step.cases: - validate_predicate(case.when, state) - profile.effect_policies[name].resolve(step.allowed_effects) + validate_predicate(case.when, state, steps=definition.spec.steps) + if isinstance(step.node, NodeTemplate): + if step.allowed_effects or step.route: + raise ValueError("user node cannot request effects or trusted routers") + for rule in step.node.rules: + validate_predicate(rule.when, state, steps=definition.spec.steps) + else: + profile.effect_policies[name].resolve(step.allowed_effects) targets = ( [step.next] if step.next @@ -63,6 +76,7 @@ async def resolve_definition( project_lookup: ProjectLookup | None = None, *, state_profile: str | None = None, + node_lookup: NodeLookup | None = None, ) -> WorkflowDefinition: """Pin all active dependencies and replace includes with their trusted steps.""" @@ -156,6 +170,23 @@ async def expand( steps, dependencies = await expand( definition, (f"root:{definition.metadata.name}",), state_profile or definition.spec.state ) + resolved_steps = dict(steps) + node_dependencies = list(dependencies) + for name, step in steps.items(): + if isinstance(step.node, NodeReference): + if node_lookup is None: + raise ValueError(f"project node '{step.node.name}' needs a node resolver") + published = await node_lookup(step.node.name) + if published is None: + raise ValueError(f"project node '{step.node.name}' is not active") + if published.metadata.name != step.node.name: + raise ValueError("project node name mismatch") + resolved_steps[name] = step.model_copy(update={"node": published.spec}) + node_dependencies.append( + f"node:project:{published.metadata.name}:{published.metadata.revision}:{published.digest}" + ) + steps = resolved_steps + dependencies = tuple(node_dependencies) resolved = definition.model_copy( update={ "spec": definition.spec.model_copy( diff --git a/src/forge/workflow/declarative/manifest.py b/src/forge/workflow/declarative/manifest.py index 20c9e9d3..498d6c26 100644 --- a/src/forge/workflow/declarative/manifest.py +++ b/src/forge/workflow/declarative/manifest.py @@ -225,7 +225,7 @@ def build_process_manifest(definition: WorkflowDefinition) -> ProcessManifest: transitions = [] for name, step in definition.spec.steps.items(): binding = profile.station_bindings.get(name) - kind = ProcessNodeKind(profile.node_kind(name)) + kind = ProcessNodeKind("operation" if step.node else profile.node_kind(name)) nodes.append( ProcessNode( name=name, @@ -334,6 +334,7 @@ def step_signature(definition: WorkflowDefinition, name: str, step: Any) -> tupl else frozenset() ) return ( + step.node.model_dump_json(by_alias=True, exclude_none=True) if step.node else None, step.next, step.route, tuple(sorted(step.branches.items())), diff --git a/src/forge/workflow/declarative/models.py b/src/forge/workflow/declarative/models.py index 452200a8..d59d092e 100644 --- a/src/forge/workflow/declarative/models.py +++ b/src/forge/workflow/declarative/models.py @@ -40,6 +40,9 @@ class WorkflowPredicate(StrictModel): fact: str | None = None op: Literal["equals", "in", "isNull"] | None = None value: Any = None + scope: Literal["current", "any", "all"] = Field( + default="current", exclude_if=lambda value: value == "current" + ) all_of: tuple[WorkflowPredicate, ...] = Field( default=(), alias="all", exclude_if=lambda value: not value ) @@ -56,13 +59,17 @@ def validate_shape(self) -> WorkflowPredicate: if forms != 1: raise ValueError("predicate requires exactly one of fact, all, any, or not") if self.fact is not None: + if self.scope != "current" and not ( + self.fact.startswith("node.") and self.fact.endswith(".outcome") + ): + raise ValueError("scope is only valid for node outcome facts") if self.op is None: raise ValueError("fact predicate requires op") if self.op != "isNull" and self.value is None: raise ValueError("predicate requires value") if self.op == "isNull" and self.value is not None: raise ValueError("isNull does not accept value") - elif self.op is not None or self.value is not None: + elif self.op is not None or self.value is not None or self.scope != "current": raise ValueError("group predicate cannot declare op or value") if len(self.all_of) > 8 or len(self.any_of) > 8: raise ValueError("predicate group may contain at most 8 items") @@ -74,7 +81,86 @@ class WorkflowCase(StrictModel): next: str +OUTCOME_RE = re.compile(r"^[a-z][a-z0-9_]{0,31}$") +SELECTORS = frozenset( + {"ticket.key", "artifact.prd", "artifact.spec", "artifact.rca", "artifact.plan"} +) + + +class DecisionRule(StrictModel): + when: WorkflowPredicate + outcome: str + + +class NodeTemplate(StrictModel): + type: Literal["decision-v1", "agent-assessment-v1"] + outcomes: tuple[str, ...] = Field(min_length=2, max_length=16) + rules: tuple[DecisionRule, ...] = () + otherwise: str | None = None + instruction: str | None = Field(default=None, max_length=4000) + inputs: tuple[str, ...] = () + + @model_validator(mode="after") + def validate_template(self) -> NodeTemplate: + if len(set(self.outcomes)) != len(self.outcomes) or any( + not OUTCOME_RE.fullmatch(x) for x in self.outcomes + ): + raise ValueError("outcomes must be unique canonical names") + if self.type == "decision-v1": + if not self.rules or len(self.rules) > 32 or self.otherwise not in self.outcomes: + raise ValueError("decision requires 1-32 rules and a declared otherwise outcome") + if any(rule.outcome not in self.outcomes for rule in self.rules): + raise ValueError("decision rule uses undeclared outcome") + if self.instruction is not None or self.inputs: + raise ValueError("decision cannot declare agent inputs") + else: + if not self.instruction or not self.instruction.strip(): + raise ValueError("agent assessment requires an instruction") + if not self.inputs or len(set(self.inputs)) != len(self.inputs): + raise ValueError("agent assessment requires unique inputs") + if any(item not in SELECTORS for item in self.inputs): + raise ValueError("agent assessment uses unknown input selector") + if self.rules or self.otherwise is not None: + raise ValueError("agent assessment cannot declare decision rules") + return self + + +class NodeReference(StrictModel): + source: Literal["project"] + name: str + + @field_validator("name") + @classmethod + def valid_name(cls, value: str) -> str: + if not WORKFLOW_NAME_RE.fullmatch(value): + raise ValueError("invalid node name") + return value + + +class NodeDefinition(StrictModel): + api_version: Literal["forge/v1"] = Field(alias="apiVersion") + kind: Literal["Node"] + metadata: WorkflowMetadata + spec: NodeTemplate + + def canonical_json(self) -> str: + return json.dumps( + self.model_dump(by_alias=True, exclude_none=True, mode="json"), + sort_keys=True, + separators=(",", ":"), + ) + + @property + def digest(self) -> str: + return hashlib.sha256(self.canonical_json().encode()).hexdigest() + + def validate_size(self) -> None: + if len(self.canonical_json().encode()) > 32768: + raise ValueError("node definition exceeds 32768 bytes") + + class WorkflowStep(StrictModel): + node: NodeTemplate | NodeReference | None = None next: str | None = None route: str | None = None branches: dict[str, str] = Field(default_factory=dict) diff --git a/src/forge/workflow/declarative/node_cli.py b/src/forge/workflow/declarative/node_cli.py new file mode 100644 index 00000000..8327d0e9 --- /dev/null +++ b/src/forge/workflow/declarative/node_cli.py @@ -0,0 +1,81 @@ +"""CLI for safe project node templates.""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from typing import Any + +import yaml + +from forge.workflow.declarative.loader import LimitedSafeLoader +from forge.workflow.declarative.models import SELECTORS, NodeDefinition +from forge.workflow.declarative.node_publication import NodePublisher, validate_node + + +def load_node(path: str) -> NodeDefinition: + raw = Path(path).read_bytes() + if len(raw) > 32768: + raise ValueError("node file exceeds 32768 bytes") + return NodeDefinition.model_validate(yaml.load(raw.decode(), Loader=LimitedSafeLoader)) + + +async def cmd_node(args: Any) -> int: + try: + action = args.node_command + if action == "catalog": + print( + json.dumps( + { + "types": ["decision-v1", "agent-assessment-v1"], + "inputs": sorted(SELECTORS), + "facts": [ + "workflow.paused", + "workflow.has_error", + "ci.status", + "pr.merged", + "review.revision_requested", + ], + }, + indent=2, + ) + ) + return 0 + if action == "validate": + definition = load_node(args.file) + validate_node(definition) + print( + f"[OK] {definition.metadata.name} revision {definition.metadata.revision} ({definition.digest})" + ) + return 0 + publisher = NodePublisher(args.project_key) + if action == "publish": + definition = load_node(args.file) + await publisher.publish(definition) + print( + f"[OK] published {definition.metadata.name} revision {definition.metadata.revision} ({definition.digest})" + ) + elif action in {"activate", "rollback"}: + definition = await getattr(publisher, action)( + args.name, args.revision, expected_digest=args.expected_active_digest + ) + print( + f"[OK] {action} {definition.metadata.name} revision {definition.metadata.revision}" + ) + elif action == "show": + definition = await publisher.active(args.name) + if definition is None: + raise ValueError("active node is unavailable") + print(json.dumps(json.loads(definition.canonical_json()), indent=2)) + elif action == "history": + definitions = await publisher.history(args.name) + print(json.dumps([json.loads(item.canonical_json()) for item in definitions], indent=2)) + elif action == "list": + print("\n".join(await publisher.list_nodes())) + else: + raise ValueError("node command is required") + return 0 + except Exception as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 diff --git a/src/forge/workflow/declarative/node_publication.py b/src/forge/workflow/declarative/node_publication.py new file mode 100644 index 00000000..8aff2645 --- /dev/null +++ b/src/forge/workflow/declarative/node_publication.py @@ -0,0 +1,186 @@ +"""Project-scoped, immutable user-node publication.""" + +from __future__ import annotations + +from typing import Any + +from forge.orchestrator.checkpointer import get_redis_client +from forge.workflow.declarative.models import NodeDefinition +from forge.workflow.declarative.predicates import validate_predicate + + +def validate_node(definition: NodeDefinition) -> None: + definition.validate_size() + if definition.spec.type == "decision-v1": + for rule in definition.spec.rules: + for state in ("feature", "bug", "task_takeover"): + validate_predicate(rule.when, state) + + +class NodePublisher: + def __init__(self, project_key: str, redis_client: Any = None) -> None: + if not project_key.strip(): + raise ValueError("project key is required") + self.project_key = project_key.upper() + self._redis = redis_client + + async def _client(self) -> Any: + if self._redis is None: + self._redis = await get_redis_client() + return self._redis + + def _key(self, name: str, revision: int) -> str: + return f"forge:node:def:{self.project_key}:{name}:{revision}" + + def _active_key(self, name: str) -> str: + return f"forge:node:active:{self.project_key}:{name}" + + async def publish(self, definition: NodeDefinition) -> None: + validate_node(definition) + redis = await self._client() + key = self._key(definition.metadata.name, definition.metadata.revision) + value = definition.canonical_json() + result = await redis.eval( + """local x=redis.call('GET',KEYS[1]) + if x and x~=ARGV[1] then return -1 end + local latest=redis.call('GET',KEYS[2]) + if not x and latest and tonumber(ARGV[2])<=tonumber(latest) then return -2 end + if not x then + redis.call('SET',KEYS[1],ARGV[1]) + redis.call('SET',KEYS[2],ARGV[2]) + end + return 1""", + 2, + key, + f"forge:node:latest:{self.project_key}:{definition.metadata.name}", + value, + str(definition.metadata.revision), + ) + if result == -1: + raise ValueError("published node revision is immutable") + if result == -2: + raise ValueError("changed node content must increment metadata.revision") + + async def get(self, name: str, revision: int) -> NodeDefinition | None: + value = await (await self._client()).get(self._key(name, revision)) + return NodeDefinition.model_validate_json(value) if value else None + + async def active(self, name: str) -> NodeDefinition | None: + value = await (await self._client()).get(self._active_key(name)) + if not value: + return None + revision = int(value.decode() if isinstance(value, bytes) else value) + return await self.get(name, revision) + + async def activate( + self, name: str, revision: int, *, expected_digest: str | None = None + ) -> NodeDefinition: + target = await self.get(name, revision) + if target is None: + raise ValueError("node revision is unavailable") + previous = await self.active(name) + if previous and previous.digest != expected_digest: + raise ValueError("active node changed concurrently") + result = await (await self._client()).eval( + """local old=redis.call('GET',KEYS[1]) + if ARGV[1]~='' and old~=ARGV[1] then return -1 end + if ARGV[1]=='' and old then return -1 end + redis.call('SET',KEYS[1],ARGV[2]); return 1""", + 1, + self._active_key(name), + str(previous.metadata.revision) if previous else "", + str(revision), + ) + if result == -1: + raise ValueError("active node changed concurrently") + return target + + async def rollback(self, name: str, revision: int, *, expected_digest: str) -> NodeDefinition: + previous = await self.active(name) + if previous is None or revision >= previous.metadata.revision: + raise ValueError("rollback requires an older revision") + return await self.activate(name, revision, expected_digest=expected_digest) + + async def history(self, name: str) -> tuple[NodeDefinition, ...]: + redis = await self._client() + keys = [] + cursor = 0 + while True: + cursor, found = await redis.scan(cursor=cursor, match=self._key(name, "*")) + keys.extend(found) + if cursor == 0: + break + result = [ + NodeDefinition.model_validate_json(value) + for key in keys + if (value := await redis.get(key)) + ] + return tuple(sorted(result, key=lambda item: item.metadata.revision)) + + async def list_nodes(self) -> tuple[str, ...]: + redis = await self._client() + names = set() + cursor = 0 + while True: + cursor, found = await redis.scan( + cursor=cursor, match=f"forge:node:def:{self.project_key}:*" + ) + for key in found: + names.add((key.decode() if isinstance(key, bytes) else key).rsplit(":", 2)[-2]) + if cursor == 0: + break + return tuple(sorted(names)) + + +class InMemoryNodePublisher: + def __init__(self, project_key: str = "DEFAULT") -> None: + self.project_key = project_key.upper() + self._definitions: dict[tuple[str, int], NodeDefinition] = {} + self._active: dict[str, NodeDefinition] = {} + + async def publish(self, definition: NodeDefinition) -> None: + validate_node(definition) + key = (definition.metadata.name, definition.metadata.revision) + if key in self._definitions and self._definitions[key].digest != definition.digest: + raise ValueError("published node revision is immutable") + if key not in self._definitions and any( + item == definition.metadata.name and rev >= definition.metadata.revision + for item, rev in self._definitions + ): + raise ValueError("changed node content must increment metadata.revision") + self._definitions[key] = definition + + async def get(self, name: str, revision: int) -> NodeDefinition | None: + return self._definitions.get((name, revision)) + + async def active(self, name: str) -> NodeDefinition | None: + return self._active.get(name) + + async def activate( + self, name: str, revision: int, *, expected_digest: str | None = None + ) -> NodeDefinition: + target = await self.get(name, revision) + if target is None: + raise ValueError("node revision is unavailable") + previous = self._active.get(name) + if previous and previous.digest != expected_digest: + raise ValueError("active node changed concurrently") + self._active[name] = target + return target + + async def rollback(self, name: str, revision: int, *, expected_digest: str) -> NodeDefinition: + previous = self._active.get(name) + if previous is None or revision >= previous.metadata.revision: + raise ValueError("rollback requires an older revision") + return await self.activate(name, revision, expected_digest=expected_digest) + + async def history(self, name: str) -> tuple[NodeDefinition, ...]: + return tuple( + sorted( + (value for (item, _), value in self._definitions.items() if item == name), + key=lambda value: value.metadata.revision, + ) + ) + + async def list_nodes(self) -> tuple[str, ...]: + return tuple(sorted({name for name, _ in self._definitions})) diff --git a/src/forge/workflow/declarative/predicates.py b/src/forge/workflow/declarative/predicates.py index d9429085..f09d1fd6 100644 --- a/src/forge/workflow/declarative/predicates.py +++ b/src/forge/workflow/declarative/predicates.py @@ -6,7 +6,7 @@ from dataclasses import dataclass from typing import Any -from forge.workflow.declarative.models import WorkflowPredicate +from forge.workflow.declarative.models import WorkflowPredicate, WorkflowStep @dataclass(frozen=True) @@ -39,7 +39,13 @@ class Fact: FACT_CONTRACT_VERSION = "1" -def validate_predicate(predicate: WorkflowPredicate, state_profile: str, depth: int = 0) -> None: +def validate_predicate( + predicate: WorkflowPredicate, + state_profile: str, + depth: int = 0, + *, + steps: dict[str, WorkflowStep] | None = None, +) -> None: if depth > 3: raise ValueError("predicate nesting exceeds 3 levels") if predicate.fact is None: @@ -47,7 +53,22 @@ def validate_predicate(predicate: WorkflowPredicate, state_profile: str, depth: if predicate.not_ is not None: children = (*children, predicate.not_) for child in children: - validate_predicate(child, state_profile, depth + 1) + validate_predicate(child, state_profile, depth + 1, steps=steps) + return + if predicate.fact.startswith("node.") and predicate.fact.endswith(".outcome"): + step_name = predicate.fact[5:-8] + template = steps.get(step_name).node if steps and step_name in steps else None + if template is None or not hasattr(template, "outcomes"): + raise ValueError(f"unknown node outcome fact '{predicate.fact}'") + if predicate.op == "isNull": + return + values = predicate.value if predicate.op == "in" else [predicate.value] + if ( + not isinstance(values, list) + or not 1 <= len(values) <= 32 + or any(value not in template.outcomes for value in values) + ): + raise ValueError(f"unknown outcome for '{predicate.fact}'") return if predicate.fact not in PROFILE_FACTS[state_profile]: raise ValueError(f"unknown conditional fact '{predicate.fact}' for state '{state_profile}'") @@ -74,7 +95,36 @@ def evaluate_predicate(predicate: WorkflowPredicate, state: Mapping[str, Any]) - if predicate.not_ is not None: return not evaluate_predicate(predicate.not_, state) assert predicate.fact is not None - value = FACTS[predicate.fact].project(state) + if predicate.fact.startswith("node.") and predicate.fact.endswith(".outcome"): + step_name = predicate.fact[5:-8] + records = state.get("node_results") or {} + if predicate.scope != "current": + total = state.get("parallel_total_branches") + if not isinstance(total, int) or total < 1: + raise ValueError("parallel branch count is unavailable") + values = [] + for branch_id in range(total): + record = records.get(f"{step_name}:{branch_id}") + if record is None: + raise ValueError(f"node outcome missing for branch {branch_id}") + values.append(record.get("outcome")) + + def matches(item: Any) -> bool: + if predicate.op == "isNull": + return item is None + if predicate.op == "equals": + return item == predicate.value + return item in predicate.value + + return ( + any(map(matches, values)) if predicate.scope == "any" else all(map(matches, values)) + ) + branch = state.get("parallel_branch_id") + key = f"{step_name}:{branch if branch is not None else 'main'}" + record = records.get(key) + value = record.get("outcome") if record else None + else: + value = FACTS[predicate.fact].project(state) if predicate.op == "isNull": return value is None if predicate.op == "equals": diff --git a/src/forge/workflow/declarative/publication.py b/src/forge/workflow/declarative/publication.py index 03703b60..5bf99c83 100644 --- a/src/forge/workflow/declarative/publication.py +++ b/src/forge/workflow/declarative/publication.py @@ -13,6 +13,7 @@ from forge.workflow.declarative.composition import resolve_definition, validate_subworkflow from forge.workflow.declarative.manifest import ProcessChangeImpact, compare_process_definitions from forge.workflow.declarative.models import WorkflowDefinition +from forge.workflow.declarative.node_publication import InMemoryNodePublisher, NodePublisher _DEFINITION_PREFIX = "forge:process:def:" _ACTIVE_PREFIX = "forge:process:active:" @@ -185,8 +186,16 @@ async def _set_active( impact = None if previous and target.kind == previous.kind == "Workflow": impact = compare_process_definitions( - await resolve_definition(previous, self.active), - await resolve_definition(target, self.active), + await resolve_definition( + previous, + self.active, + node_lookup=NodePublisher(self.project_key, self._redis).active, + ), + await resolve_definition( + target, + self.active, + node_lookup=NodePublisher(self.project_key, self._redis).active, + ), ) if impact is not None and not _compatible_impact(impact, rollback=action == "rollback"): raise ValueError("definition is incompatible with active workflow instances") @@ -262,10 +271,19 @@ async def _validate(self, definition: WorkflowDefinition) -> None: if definition.kind == "Subworkflow": for state in definition.spec.compatible_states or (definition.spec.state,): validate_subworkflow( - await resolve_definition(definition, self.active, state_profile=state) + await resolve_definition( + definition, + self.active, + state_profile=state, + node_lookup=NodePublisher(self.project_key, self._redis).active, + ) ) else: - resolved = await resolve_definition(definition, self.active) + resolved = await resolve_definition( + definition, + self.active, + node_lookup=NodePublisher(self.project_key, self._redis).active, + ) DeclarativeWorkflowCompiler(resolved).validate_for_publication() async def _validate_consumers(self, name: str, candidate: WorkflowDefinition) -> None: @@ -276,7 +294,9 @@ async def lookup(dependency: str) -> WorkflowDefinition | None: consumer = await self.active(consumer_name) if consumer is None or consumer_name == name or consumer.kind != "Workflow": continue - resolved = await resolve_definition(consumer, lookup) + resolved = await resolve_definition( + consumer, lookup, node_lookup=NodePublisher(self.project_key, self._redis).active + ) DeclarativeWorkflowCompiler(resolved).validate_for_publication() def _decision( @@ -339,14 +359,20 @@ def _pointer(definition: WorkflowDefinition) -> str: class InMemoryDefinitionPublisher: """Deterministic project-scoped publisher for local and contract tests.""" - def __init__(self, project_key: str = "DEFAULT") -> None: + def __init__( + self, project_key: str = "DEFAULT", node_publisher: InMemoryNodePublisher | None = None + ) -> None: if not project_key or not project_key.strip(): raise ValueError("project_key is required for governed publication") self.project_key = project_key.upper() + self.node_publisher = node_publisher or InMemoryNodePublisher(self.project_key) self._definitions: dict[tuple[str, int], WorkflowDefinition] = {} self._active: dict[str, WorkflowDefinition] = {} self._decisions: dict[str, list[PublicationDecision]] = {} + async def node_active(self, name: str): + return await self.node_publisher.active(name) + async def publish( self, definition: WorkflowDefinition, *, actor: str, reason: str, activate: bool = False ) -> PublicationDecision: @@ -437,8 +463,12 @@ async def _set_active( impact = None if current and target.kind == current.kind == "Workflow": impact = compare_process_definitions( - await resolve_definition(current, self.active), - await resolve_definition(target, self.active), + await resolve_definition( + current, self.active, node_lookup=self.node_publisher.active + ), + await resolve_definition( + target, self.active, node_lookup=self.node_publisher.active + ), ) if impact and not _compatible_impact(impact, rollback=action == "rollback"): raise ValueError("definition is incompatible with active workflow instances") @@ -474,10 +504,17 @@ async def _validate(self, definition: WorkflowDefinition) -> None: if definition.kind == "Subworkflow": for state in definition.spec.compatible_states or (definition.spec.state,): validate_subworkflow( - await resolve_definition(definition, self.active, state_profile=state) + await resolve_definition( + definition, + self.active, + state_profile=state, + node_lookup=self.node_publisher.active, + ) ) else: - resolved = await resolve_definition(definition, self.active) + resolved = await resolve_definition( + definition, self.active, node_lookup=self.node_publisher.active + ) DeclarativeWorkflowCompiler(resolved).validate_for_publication() async def _validate_consumers(self, name: str, candidate: WorkflowDefinition) -> None: @@ -488,7 +525,9 @@ async def lookup(dependency: str) -> WorkflowDefinition | None: consumer = await self.active(consumer_name) if consumer is None or consumer_name == name or consumer.kind != "Workflow": continue - resolved = await resolve_definition(consumer, lookup) + resolved = await resolve_definition( + consumer, lookup, node_lookup=self.node_publisher.active + ) DeclarativeWorkflowCompiler(resolved).validate_for_publication() def _decision( diff --git a/src/forge/workflow/declarative/resolver.py b/src/forge/workflow/declarative/resolver.py index c87c5d8b..c26f7b46 100644 --- a/src/forge/workflow/declarative/resolver.py +++ b/src/forge/workflow/declarative/resolver.py @@ -11,6 +11,7 @@ WORKFLOW_NAME_RE, WORKFLOW_PROPERTY_PREFIX, ) +from forge.workflow.declarative.node_publication import NodePublisher from forge.workflow.declarative.workflow import DeclarativeWorkflow @@ -101,7 +102,10 @@ async def load_project_workflow( f"project {project_key.upper()} does not define workflow '{workflow_name}'" ) definition = load_workflow_value(value) - if definition.spec.includes: + if definition.spec.includes or any( + step.node is not None and getattr(step.node, "source", None) == "project" + for step in definition.spec.steps.values() + ): loaded_dependencies: dict[str, str] = {} dependency_cache: dict[str, Any] = {} @@ -123,7 +127,26 @@ async def dependency(name: str): dependency_cache[name] = resolved return resolved - definition = await resolve_definition(definition, dependency) + node_reader = ( + definition_reader.node_active + if definition_reader is not None and hasattr(definition_reader, "node_active") + else NodePublisher(project_key).active + ) + loaded_nodes: dict[str, str] = {} + + async def node_dependency(name: str): + node = await node_reader(name) + if node is not None: + loaded_nodes[name] = node.digest + return node + + definition = await resolve_definition( + definition, dependency, node_lookup=node_dependency + ) + for name, digest in loaded_nodes.items(): + current_node = await node_reader(name) + if current_node is None or current_node.digest != digest: + raise ValueError(f"project node '{name}' changed during resolution") for name, digest in loaded_dependencies.items(): current = await definition_reader.active(name) if definition_reader else None if current is None: diff --git a/src/forge/workflow/declarative/user_nodes.py b/src/forge/workflow/declarative/user_nodes.py new file mode 100644 index 00000000..82c1519e --- /dev/null +++ b/src/forge/workflow/declarative/user_nodes.py @@ -0,0 +1,75 @@ +"""Safe, versioned templates for project-authored workflow nodes.""" + +from __future__ import annotations + +import json +from typing import Any, Literal + +from pydantic import create_model + +from forge.workflow.declarative.models import NodeTemplate + + +def select_inputs(template: NodeTemplate, state: dict[str, Any]) -> dict[str, str]: + selected: dict[str, str] = {} + fields = { + "artifact.prd": "prd_content", + "artifact.spec": "spec_content", + "artifact.rca": "rca_content", + "artifact.plan": "plan_content", + } + for selector in template.inputs: + value = state.get("ticket_key") if selector == "ticket.key" else state.get(fields[selector]) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"required input '{selector}' is unavailable") + selected[selector] = value[:16000] + return selected + + +async def execute_node(template: NodeTemplate, state: dict[str, Any], step: str) -> dict[str, Any]: + from forge.workflow.declarative.predicates import evaluate_predicate + + if template.type == "decision-v1": + outcome = next( + (rule.outcome for rule in template.rules if evaluate_predicate(rule.when, state)), + template.otherwise, + ) + summary = None + else: + from forge.config import get_settings + from forge.integrations.agents.agent import ForgeAgent + from forge.model_policy import resolve_model_target_for_project + + inputs = select_inputs(template, state) + project = str( + state.get("workflow_project_key") or state.get("ticket_key", "").split("-", 1)[0] + ).upper() + settings = get_settings() + target = await resolve_model_target_for_project(settings, project, step) + model = ForgeAgent(settings)._create_model(model_target=target) + Response = create_model( + "NodeAssessment", outcome=(Literal[tuple(template.outcomes)], ...), summary=(str, ...) + ) + prompt = json.dumps( + { + "instruction": template.instruction, + "inputs": inputs, + "allowed_outcomes": template.outcomes, + } + ) + response = await model.with_structured_output(Response).ainvoke(prompt) + parsed = Response.model_validate(response) + outcome, summary = parsed.outcome, parsed.summary + if len(summary) > 1000: + raise ValueError("assessment summary exceeds 1000 characters") + branch = state.get("parallel_branch_id") + attempts = state.get("workflow_node_attempts") or {} + key = f"{step}:{branch if branch is not None else 'main'}" + record = { + "step": step, + "branch": branch, + "attempt": int(attempts.get(step, 0)) + 1, + "outcome": outcome, + "summary": summary, + } + return {**state, "node_results": {key: record}} diff --git a/tests/unit/workflow/test_user_nodes.py b/tests/unit/workflow/test_user_nodes.py new file mode 100644 index 00000000..38496247 --- /dev/null +++ b/tests/unit/workflow/test_user_nodes.py @@ -0,0 +1,234 @@ +"""Project node templates are pinned, typed and side-effect free.""" + +from __future__ import annotations + +import pytest + +from forge.workflow.base import merge_node_results +from forge.workflow.declarative.compiler import DeclarativeWorkflowCompiler, WorkflowValidationError +from forge.workflow.declarative.composition import resolve_definition +from forge.workflow.declarative.models import NodeDefinition, NodeTemplate, WorkflowDefinition +from forge.workflow.declarative.node_publication import InMemoryNodePublisher +from forge.workflow.declarative.predicates import evaluate_predicate +from forge.workflow.declarative.user_nodes import execute_node, select_inputs + + +def decision(otherwise: str = "revise") -> dict: + return { + "type": "decision-v1", + "outcomes": ["ready", "revise"], + "rules": [ + { + "when": {"fact": "workflow.has_error", "op": "equals", "value": False}, + "outcome": "ready", + } + ], + "otherwise": otherwise, + } + + +def workflow(node: dict) -> WorkflowDefinition: + return WorkflowDefinition.model_validate( + { + "apiVersion": "forge/v1", + "kind": "Workflow", + "metadata": {"name": "artifact-review", "revision": 1}, + "spec": { + "state": "feature", + "entry": "review_artifact", + "steps": { + "review_artifact": { + "node": node, + "cases": [ + { + "when": { + "fact": "node.review_artifact.outcome", + "op": "equals", + "value": "ready", + }, + "next": "generate_prd", + } + ], + "otherwise": "__end__", + }, + "generate_prd": {"next": "__end__"}, + }, + }, + } + ) + + +@pytest.mark.asyncio +async def test_inline_decision_result_routes_and_records_attempt() -> None: + definition = workflow(decision()) + compiler = DeclarativeWorkflowCompiler(definition) + compiler.build_graph().compile() + result = await execute_node( + definition.spec.steps["review_artifact"].node, + { + "ticket_key": "TEST-1", + "last_error": None, + "workflow_node_attempts": {"review_artifact": 2}, + }, + "review_artifact", + ) + assert result["node_results"]["review_artifact:main"]["attempt"] == 3 + assert result["node_results"]["review_artifact:main"]["outcome"] == "ready" + assert ( + compiler._conditional_route(definition.spec.steps["review_artifact"].cases, "__end__")( + result + ) + == "generate_prd" + ) + + +@pytest.mark.asyncio +async def test_project_reference_pins_active_revision() -> None: + publisher = InMemoryNodePublisher("TEST") + first = NodeDefinition.model_validate( + { + "apiVersion": "forge/v1", + "kind": "Node", + "metadata": {"name": "review", "revision": 1}, + "spec": decision(), + } + ) + await publisher.publish(first) + await publisher.activate("review", 1) + source = workflow({"source": "project", "name": "review"}) + pinned = await resolve_definition(source, node_lookup=publisher.active) + assert pinned.spec.steps["review_artifact"].node.otherwise == "revise" + second = first.model_copy( + update={ + "metadata": first.metadata.model_copy(update={"revision": 2}), + "spec": NodeTemplate.model_validate( + { + **decision(), + "rules": [ + { + "when": {"fact": "workflow.has_error", "op": "equals", "value": False}, + "outcome": "revise", + } + ], + } + ), + } + ) + await publisher.publish(second) + await publisher.activate("review", 2, expected_digest=first.digest) + assert (await resolve_definition(source, node_lookup=publisher.active)).digest != pinned.digest + assert (await resolve_definition(pinned, node_lookup=publisher.active)).digest == pinned.digest + + +def test_unknown_outcome_and_missing_input_fail_closed() -> None: + definition = workflow(decision()) + bad = definition.model_copy( + update={ + "spec": definition.spec.model_copy( + update={ + "steps": { + **definition.spec.steps, + "review_artifact": definition.spec.steps["review_artifact"].model_copy( + update={ + "cases": ( + definition.spec.steps["review_artifact"] + .cases[0] + .model_copy( + update={ + "when": definition.spec.steps["review_artifact"] + .cases[0] + .when.model_copy(update={"value": "unknown"}) + } + ), + ) + } + ), + } + } + ) + } + ) + with pytest.raises(WorkflowValidationError, match="unknown outcome"): + DeclarativeWorkflowCompiler(bad).validate() + template = NodeTemplate( + type="agent-assessment-v1", + outcomes=("ready", "revise"), + instruction="Assess artifact", + inputs=("artifact.spec",), + ) + with pytest.raises(ValueError, match="unavailable"): + select_inputs(template, {"ticket_key": "TEST-1"}) + + +def test_parallel_results_merge_and_scope() -> None: + left = {"review_artifact:0": {"attempt": 1, "outcome": "ready"}} + right = {"review_artifact:1": {"attempt": 1, "outcome": "revise"}} + merged = merge_node_results(left, right) + predicate = ( + workflow(decision()) + .spec.steps["review_artifact"] + .cases[0] + .when.model_copy(update={"scope": "any"}) + ) + assert evaluate_predicate(predicate, {"node_results": merged, "parallel_total_branches": 2}) + assert not evaluate_predicate( + predicate.model_copy(update={"scope": "all"}), + {"node_results": merged, "parallel_total_branches": 2}, + ) + with pytest.raises(ValueError, match="missing"): + evaluate_predicate(predicate, {"node_results": left, "parallel_total_branches": 2}) + + +@pytest.mark.asyncio +async def test_agent_assessment_receives_only_selected_inputs( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import json + + import forge.config + import forge.integrations.agents.agent + import forge.model_policy + + prompts = [] + + class FakeModel: + def with_structured_output(self, schema): + self.schema = schema + return self + + async def ainvoke(self, prompt): + prompts.append(json.loads(prompt)) + return {"outcome": "ready", "summary": "Acceptance criteria are clear."} + + class FakeAgent: + def __init__(self, settings): + pass + + def _create_model(self, *, model_target): + _ = model_target + return FakeModel() + + async def target(*args): + _ = args + return object() + + monkeypatch.setattr(forge.config, "get_settings", lambda: object()) + monkeypatch.setattr(forge.model_policy, "resolve_model_target_for_project", target) + monkeypatch.setattr(forge.integrations.agents.agent, "ForgeAgent", FakeAgent) + template = NodeTemplate( + type="agent-assessment-v1", + outcomes=("ready", "revise"), + instruction="Review the specification", + inputs=("artifact.spec",), + ) + result = await execute_node( + template, + {"ticket_key": "DEMO-1", "spec_content": "Clear requirements", "secret": "private"}, + "review_artifact", + ) + assert prompts[0]["inputs"] == {"artifact.spec": "Clear requirements"} + assert "private" not in json.dumps(prompts) + assert ( + result["node_results"]["review_artifact:main"]["summary"] + == "Acceptance criteria are clear." + ) From ec6743fb444b83ab40497da958b74e7011eef31d Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Wed, 30 Sep 2026 16:58:31 +0300 Subject: [PATCH 2/2] fix: preserve user-node policy, retry, and consumer contracts --- docs/examples/workflow-nodes/README.md | 8 + docs/reference/config.md | 4 +- src/forge/models/model_policy.py | 3 + src/forge/workflow/declarative/compiler.py | 1 + .../workflow/declarative/node_publication.py | 52 +++- src/forge/workflow/declarative/publication.py | 1 + src/forge/workflow/declarative/user_nodes.py | 11 +- tests/unit/models/test_model_policy.py | 1 + tests/unit/workflow/test_user_nodes.py | 241 +++++++++++++++++- 9 files changed, 315 insertions(+), 7 deletions(-) diff --git a/docs/examples/workflow-nodes/README.md b/docs/examples/workflow-nodes/README.md index f0cf38cc..7d7b0975 100644 --- a/docs/examples/workflow-nodes/README.md +++ b/docs/examples/workflow-nodes/README.md @@ -17,6 +17,10 @@ review_artifact: 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 @@ -27,4 +31,8 @@ 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 ` 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..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. diff --git a/docs/reference/config.md b/docs/reference/config.md index 5428bbab..d7320dd1 100644 --- a/docs/reference/config.md +++ b/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 diff --git a/src/forge/models/model_policy.py b/src/forge/models/model_policy.py index e85eff0c..c703c065 100644 --- a/src/forge/models/model_policy.py +++ b/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( diff --git a/src/forge/workflow/declarative/compiler.py b/src/forge/workflow/declarative/compiler.py index 2171548f..3dc113c9 100644 --- a/src/forge/workflow/declarative/compiler.py +++ b/src/forge/workflow/declarative/compiler.py @@ -374,6 +374,7 @@ async def run(state: dict[str, Any]) -> dict[str, Any]: except Exception as exc: return { **state, + "current_node": node_name, "is_blocked": True, "last_error": f"User node {node_name}: {exc}", } diff --git a/src/forge/workflow/declarative/node_publication.py b/src/forge/workflow/declarative/node_publication.py index 8aff2645..bfca56a8 100644 --- a/src/forge/workflow/declarative/node_publication.py +++ b/src/forge/workflow/declarative/node_publication.py @@ -2,10 +2,16 @@ from __future__ import annotations -from typing import Any +from typing import Any, Protocol from forge.orchestrator.checkpointer import get_redis_client -from forge.workflow.declarative.models import NodeDefinition +from forge.workflow.declarative.compiler import DeclarativeWorkflowCompiler +from forge.workflow.declarative.composition import ( + NodeLookup, + resolve_definition, + validate_subworkflow, +) +from forge.workflow.declarative.models import NodeDefinition, WorkflowDefinition from forge.workflow.declarative.predicates import validate_predicate @@ -17,6 +23,41 @@ def validate_node(definition: NodeDefinition) -> None: validate_predicate(rule.when, state) +class WorkflowReader(Protocol): + async def list_workflows(self) -> tuple[str, ...]: ... + + async def active(self, name: str) -> WorkflowDefinition | None: ... + + +async def _validate_consumers( + candidate: NodeDefinition, workflows: WorkflowReader, node_lookup: NodeLookup +) -> None: + """Check active consumers with the candidate substituted before changing the pointer.""" + + async def lookup(name: str) -> NodeDefinition | None: + return candidate if name == candidate.metadata.name else await node_lookup(name) + + for name in await workflows.list_workflows(): + consumer = await workflows.active(name) + if consumer is None: + continue + try: + if consumer.kind == "Subworkflow": + for state in consumer.spec.compatible_states or (consumer.spec.state,): + validate_subworkflow( + await resolve_definition( + consumer, workflows.active, state_profile=state, node_lookup=lookup + ) + ) + else: + resolved = await resolve_definition(consumer, workflows.active, node_lookup=lookup) + DeclarativeWorkflowCompiler(resolved).validate_for_publication() + except ValueError as exc: + raise ValueError( + f"node '{candidate.metadata.name}' invalidates active consumer '{name}': {exc}" + ) from exc + + class NodePublisher: def __init__(self, project_key: str, redis_client: Any = None) -> None: if not project_key.strip(): @@ -81,6 +122,10 @@ async def activate( previous = await self.active(name) if previous and previous.digest != expected_digest: raise ValueError("active node changed concurrently") + from forge.workflow.declarative.publication import DefinitionPublisher + + workflows = DefinitionPublisher(self.project_key, await self._client()) + await _validate_consumers(target, workflows, self.active) result = await (await self._client()).eval( """local old=redis.call('GET',KEYS[1]) if ARGV[1]~='' and old~=ARGV[1] then return -1 end @@ -137,6 +182,7 @@ def __init__(self, project_key: str = "DEFAULT") -> None: self.project_key = project_key.upper() self._definitions: dict[tuple[str, int], NodeDefinition] = {} self._active: dict[str, NodeDefinition] = {} + self.workflow_reader: WorkflowReader | None = None async def publish(self, definition: NodeDefinition) -> None: validate_node(definition) @@ -165,6 +211,8 @@ async def activate( previous = self._active.get(name) if previous and previous.digest != expected_digest: raise ValueError("active node changed concurrently") + if self.workflow_reader is not None: + await _validate_consumers(target, self.workflow_reader, self.active) self._active[name] = target return target diff --git a/src/forge/workflow/declarative/publication.py b/src/forge/workflow/declarative/publication.py index 5bf99c83..498e6091 100644 --- a/src/forge/workflow/declarative/publication.py +++ b/src/forge/workflow/declarative/publication.py @@ -366,6 +366,7 @@ def __init__( raise ValueError("project_key is required for governed publication") self.project_key = project_key.upper() self.node_publisher = node_publisher or InMemoryNodePublisher(self.project_key) + self.node_publisher.workflow_reader = self self._definitions: dict[tuple[str, int], WorkflowDefinition] = {} self._active: dict[str, WorkflowDefinition] = {} self._decisions: dict[str, list[PublicationDecision]] = {} diff --git a/src/forge/workflow/declarative/user_nodes.py b/src/forge/workflow/declarative/user_nodes.py index 82c1519e..01a86bff 100644 --- a/src/forge/workflow/declarative/user_nodes.py +++ b/src/forge/workflow/declarative/user_nodes.py @@ -7,6 +7,7 @@ from pydantic import create_model +from forge.workflow.base import merge_node_results from forge.workflow.declarative.models import NodeTemplate @@ -45,7 +46,7 @@ async def execute_node(template: NodeTemplate, state: dict[str, Any], step: str) state.get("workflow_project_key") or state.get("ticket_key", "").split("-", 1)[0] ).upper() settings = get_settings() - target = await resolve_model_target_for_project(settings, project, step) + target = await resolve_model_target_for_project(settings, project, "user_node_assessment") model = ForgeAgent(settings)._create_model(model_target=target) Response = create_model( "NodeAssessment", outcome=(Literal[tuple(template.outcomes)], ...), summary=(str, ...) @@ -72,4 +73,10 @@ async def execute_node(template: NodeTemplate, state: dict[str, Any], step: str) "outcome": outcome, "summary": summary, } - return {**state, "node_results": {key: record}} + # The compiler evaluates outgoing cases before LangGraph applies reducers. + # Supply the complete outcome view here as well as in the checkpoint. + return { + **state, + "current_node": step, + "node_results": merge_node_results(state.get("node_results"), {key: record}), + } diff --git a/tests/unit/models/test_model_policy.py b/tests/unit/models/test_model_policy.py index 93106563..dc5f099d 100644 --- a/tests/unit/models/test_model_policy.py +++ b/tests/unit/models/test_model_policy.py @@ -110,6 +110,7 @@ def test_tool_requirements_cover_every_agentic_stage() -> None: "sync_pr_description", } assert REQUIRED_CAPABILITIES_BY_POLICY_KEY["automated_review_triage"] == {"structured_output"} + assert REQUIRED_CAPABILITIES_BY_POLICY_KEY["user_node_assessment"] == {"structured_output"} def test_project_output_token_limit_is_bounded(resolver: ModelPolicyResolver) -> None: diff --git a/tests/unit/workflow/test_user_nodes.py b/tests/unit/workflow/test_user_nodes.py index 38496247..7f5ffcf2 100644 --- a/tests/unit/workflow/test_user_nodes.py +++ b/tests/unit/workflow/test_user_nodes.py @@ -2,14 +2,27 @@ from __future__ import annotations +from unittest.mock import AsyncMock + import pytest +from forge.orchestrator.command_handlers import _apply_retry from forge.workflow.base import merge_node_results +from forge.workflow.declarative.builtins import builtin_feature_definition from forge.workflow.declarative.compiler import DeclarativeWorkflowCompiler, WorkflowValidationError from forge.workflow.declarative.composition import resolve_definition -from forge.workflow.declarative.models import NodeDefinition, NodeTemplate, WorkflowDefinition -from forge.workflow.declarative.node_publication import InMemoryNodePublisher +from forge.workflow.declarative.models import ( + NodeDefinition, + NodeTemplate, + WorkflowCase, + WorkflowDefinition, + WorkflowInclude, + WorkflowPredicate, +) +from forge.workflow.declarative.node_publication import InMemoryNodePublisher, NodePublisher from forge.workflow.declarative.predicates import evaluate_predicate +from forge.workflow.declarative.publication import InMemoryDefinitionPublisher +from forge.workflow.declarative.resolver import load_project_workflow from forge.workflow.declarative.user_nodes import execute_node, select_inputs @@ -232,3 +245,227 @@ async def target(*args): result["node_results"]["review_artifact:main"]["summary"] == "Acceptance criteria are clear." ) + + +@pytest.mark.asyncio +async def test_assessment_resolves_explicit_model_policy(mock_settings, monkeypatch) -> None: + import forge.config + from forge.integrations.agents.agent import ForgeAgent + + settings = mock_settings.model_copy( + update={"model_default": {"connection": "default", "model": mock_settings.llm_model}} + ) + targets = [] + + class Model: + def with_structured_output(self, _schema): + return self + + async def ainvoke(self, _prompt): + return {"outcome": "ready", "summary": "Clear requirements."} + + def create_model(_self, *, model_target): + targets.append(model_target) + return Model() + + monkeypatch.setattr(forge.config, "get_settings", lambda: settings) + monkeypatch.setattr(ForgeAgent, "_create_model", create_model) + template = NodeTemplate( + type="agent-assessment-v1", + outcomes=("ready", "revise"), + instruction="Assess the spec", + inputs=("artifact.spec",), + ) + result = await execute_node( + template, {"ticket_key": "TEST-1", "spec_content": "Spec"}, "custom_assessment" + ) + assert targets[0].policy_key == "user_node_assessment" + assert targets[0].model == mock_settings.llm_model + assert result["node_results"]["custom_assessment:main"]["outcome"] == "ready" + + +@pytest.mark.asyncio +async def test_failed_assessment_retry_returns_to_failed_node() -> None: + definition = workflow( + { + "type": "agent-assessment-v1", + "outcomes": ["ready", "revise"], + "instruction": "Assess the spec", + "inputs": ["artifact.spec"], + } + ) + compiler = DeclarativeWorkflowCompiler(definition) + failed = await compiler._guarded_node( + compiler._node_function("review_artifact"), "review_artifact", terminal=False + )({"ticket_key": "TEST-1", "current_node": "generate_prd"}) + assert failed["is_blocked"] + assert "required input" in failed["last_error"] + retry = _apply_retry(None, failed).state + assert not retry["is_blocked"] + assert compiler._entry_route()(retry) == "review_artifact" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("scope", ["current", "any", "all"]) +async def test_user_node_conditions_see_prior_outcomes(scope: str) -> None: + template = NodeTemplate.model_validate(decision()) + records = { + f"review_artifact:{branch}": {"attempt": 1, "outcome": "ready"} for branch in ("main", 0, 1) + } + state = {"node_results": records, "parallel_total_branches": 2} + case = WorkflowCase( + when=WorkflowPredicate( + fact="node.review_artifact.outcome", scope=scope, op="equals", value="ready" + ), + next="generate_prd", + ) + + async def run(value): + return await execute_node(template, value, "aggregate") + + compiler = DeclarativeWorkflowCompiler(workflow(decision())) + result = await compiler._guarded_node( + run, "aggregate", terminal=False, cases=(case,), otherwise="__end__" + )(state) + assert result["transition_history"][-1]["target"] == "generate_prd" + assert compiler._conditional_route((case,), "__end__")(result) == "generate_prd" + assert result["node_results"]["aggregate:main"]["outcome"] == "ready" + assert all(result["node_results"][key] == value for key, value in records.items()) + + +def _node_revision(revision: int, ready: str = "ready", revise: str = "revise") -> NodeDefinition: + spec = decision() + spec["outcomes"] = [ready, revise] + spec["rules"][0]["outcome"] = ready + spec["otherwise"] = revise + return NodeDefinition.model_validate( + { + "apiVersion": "forge/v1", + "kind": "Node", + "metadata": {"name": "review", "revision": revision}, + "spec": spec, + } + ) + + +async def _activate_consumer(publisher, *, nested: bool) -> None: + original = builtin_feature_definition() + review_step = workflow({"source": "project", "name": "review"}).spec.steps["review_artifact"] + steps = dict(original.spec.steps) + includes = () + if nested: + fragment = WorkflowDefinition.model_validate( + { + "apiVersion": "forge/v1", + "kind": "Subworkflow", + "metadata": {"name": "review_fragment", "revision": 1}, + "spec": { + "state": "feature", + "entry": "review_artifact", + "steps": { + "review_artifact": { + **review_step.model_dump(by_alias=True, exclude_none=True), + "cases": [ + { + "when": review_step.cases[0].when.model_dump(by_alias=True), + "next": "@exit/ready", + } + ], + "otherwise": "@exit/revise", + } + }, + }, + } + ) + await publisher.publish(fragment, actor="test", reason="shared review") + await publisher.activate(fragment, actor="test", reason="shared review") + includes = ( + WorkflowInclude( + source="project", + name="review_fragment", + exits={"ready": "generate_prd", "revise": "__end__"}, + ), + ) + else: + steps["review_artifact"] = review_step + consumer = original.model_copy( + update={ + "metadata": original.metadata.model_copy(update={"name": "consumer", "revision": 1}), + "spec": original.spec.model_copy( + update={ + "entry": "review_artifact", + "steps": steps, + "includes": includes, + "resolved_dependencies": (), + } + ), + } + ) + await publisher.publish(consumer, actor="test", reason="review artifacts") + await publisher.activate(consumer, actor="test", reason="review artifacts") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("nested", [False, True]) +@pytest.mark.parametrize("action", ["activate", "rollback"]) +async def test_node_revision_rejects_incompatible_consumers(nested: bool, action: str) -> None: + nodes = InMemoryNodePublisher("TEST") + publisher = InMemoryDefinitionPublisher("TEST", node_publisher=nodes) + active = _node_revision(1 if action == "activate" else 2) + incompatible = _node_revision(2 if action == "activate" else 1, "pass", "fail") + for revision in sorted((active, incompatible), key=lambda item: item.metadata.revision): + await nodes.publish(revision) + await nodes.activate("review", active.metadata.revision) + await _activate_consumer(publisher, nested=nested) + + with pytest.raises(ValueError, match="invalidates active consumer.*unknown outcome"): + await getattr(nodes, action)( + "review", incompatible.metadata.revision, expected_digest=active.digest + ) + assert (await nodes.active("review")).digest == active.digest + await load_project_workflow(None, "TEST", "consumer", definition_reader=publisher) + + +@pytest.mark.asyncio +async def test_compatible_node_activation_preserves_existing_pin() -> None: + nodes = InMemoryNodePublisher("TEST") + publisher = InMemoryDefinitionPublisher("TEST", node_publisher=nodes) + first, second = _node_revision(1), _node_revision(2) + await nodes.publish(first) + await nodes.publish(second) + await nodes.activate("review", 1) + await _activate_consumer(publisher, nested=True) + pinned = await load_project_workflow(None, "TEST", "consumer", definition_reader=publisher) + await nodes.activate("review", 2, expected_digest=first.digest) + latest = await load_project_workflow(None, "TEST", "consumer", definition_reader=publisher) + assert latest.definition.digest != pinned.definition.digest + assert ( + await resolve_definition(pinned.definition, node_lookup=nodes.active) + ).digest == pinned.definition.digest + await nodes.rollback("review", 1, expected_digest=second.digest) + assert (await nodes.active("review")).digest == first.digest + + +@pytest.mark.asyncio +async def test_redis_node_activation_checks_consumers_before_pointer_write() -> None: + """Exercise the Redis publisher's validation wiring without mocking the validator.""" + first, second = _node_revision(1), _node_revision(2, "pass", "fail") + nodes = InMemoryNodePublisher("TEST") + publisher = InMemoryDefinitionPublisher("TEST", node_publisher=nodes) + await nodes.publish(first) + await nodes.activate("review", 1) + await _activate_consumer(publisher, nested=False) + consumer = await publisher.active("consumer") + data = { + "forge:node:def:TEST:review:1": first.canonical_json(), + "forge:node:def:TEST:review:2": second.canonical_json(), + "forge:node:active:TEST:review": "1", + "forge:process:def:TEST:consumer:1": consumer.canonical_json(), + "forge:process:active:TEST:consumer": f"1:{consumer.digest}", + } + redis = AsyncMock() + redis.get.side_effect = data.get + redis.scan.return_value = (0, ["forge:process:def:TEST:consumer:1"]) + with pytest.raises(ValueError, match="invalidates active consumer.*unknown outcome"): + await NodePublisher("TEST", redis).activate("review", 2, expected_digest=first.digest) + redis.eval.assert_not_awaited()