diff --git a/.agents/skills/forge-workflow-authoring/SKILL.md b/.agents/skills/forge-workflow-authoring/SKILL.md index 53ab716a..bfeeb1a7 100644 --- a/.agents/skills/forge-workflow-authoring/SKILL.md +++ b/.agents/skills/forge-workflow-authoring/SKILL.md @@ -10,10 +10,12 @@ Help the user express a Forge process as readable YAML. Treat canonical JSON as ## Start here 1. Read [references/workflow-format.md](references/workflow-format.md). -2. For a new workflow, copy [assets/workflow.yaml](assets/workflow.yaml). For a change, start from the active definition or the closest built-in workflow and convert it to YAML if needed. +2. For a small local example, copy [assets/workflow.yaml](assets/workflow.yaml). For a publishable workflow, start from the closest full built-in definition and convert it to YAML if needed; publication requires the state profile's mandatory gates. 3. Establish the intended stages, decisions, loops, human pauses, and external commands before editing fields. 4. Run `forge workflow catalog STATE` and use only the nodes and routers it reports. Never invent catalog names. -5. Keep the definition flow-only. Do not add node kinds, station contracts, effect capabilities, required or mandatory policies, extension declarations, observation policies, or external-entry flags. Forge derives and enforces those concerns from its trusted catalog and publication policy. + For reusable steps, inspect the built-in `github_pr_review` subworkflow or an active + project definition. Bind every `@exit/name` in `spec.includes` and keep node names unique. +5. Keep the definition flow-only. Do not add step kinds, station contracts, effect capabilities, required or mandatory policies, extension declarations, observation policies, or external-entry flags. Forge derives and enforces those concerns from its trusted catalog and publication policy. 6. Validate and render before presenting the result: ```bash @@ -21,6 +23,9 @@ Help the user express a Forge process as readable YAML. Treat canonical JSON as forge workflow render WORKFLOW.yaml ``` + Add `--project-key PROJECT` to these commands when the source includes active project + definitions. Built-in includes resolve locally. Render a subworkflow to inspect its exits. + Explain the rendered process in plain language when the user is trying to understand an existing definition. ## Changing an existing workflow @@ -43,11 +48,14 @@ Do not claim migration safety based only on successful validation. Before publication, verify: -- all transitions and router outcomes resolve to existing steps or `__end__`; +- workflow transitions resolve to existing steps or `__end__`; subworkflow exits use + `@exit/name` and every caller binds them; - expected human and CI pause points remain present; - cycles cross an approved pause boundary; - exceptional commands such as PR rebasing are absent from graph topology; - revision and resume mappings protect in-flight instances. +- dependency references resolve to active definitions in the same Jira project or to built-ins, + and the expanded graph has no node-name collisions. Report review problems with the affected step and a concrete correction. Distinguish topology, execution-policy, and migration findings; catalog and governance concerns are Forge implementation findings, not fields to add to the workflow. diff --git a/.agents/skills/forge-workflow-authoring/references/workflow-format.md b/.agents/skills/forge-workflow-authoring/references/workflow-format.md index 1a416ce5..b5e8df3e 100644 --- a/.agents/skills/forge-workflow-authoring/references/workflow-format.md +++ b/.agents/skills/forge-workflow-authoring/references/workflow-format.md @@ -9,7 +9,14 @@ A definition has four important parts: - `spec.entry` names the first ordinary step. - `spec.steps` maps registered node names to their transitions. -Start at `entry`. Follow `next` for a fixed transition. At a step with `route`, follow every target in `branches`. A branch key is a possible router result; its value is the next step. `__end__` stops the current invocation and is not itself a declared step. +Reusable `Subworkflow` definitions use the same registered nodes and may end a transition +at `@exit/name`. A caller binds each exit through `spec.includes`; it may also include a +complete workflow with `returnTo` and `returnFrom` for selected normal completion edges. Use `source: builtin` or +`source: project`. Project references resolve to active definitions in the caller's Jira +project. Nodes remain canonical and cannot collide. New runs pin the fully expanded graph; +active runs keep their existing pin. + +Start at `entry`. Follow `next` for a fixed transition. At a step with `route`, follow every target in `branches`. A branch key is a possible router result; its value is the next step. `__end__` stops the current invocation and is not itself a declared step. In an included full workflow, `returnFrom` lists the terminal edges that return to the caller's `returnTo` target. Write `step_name` for a fixed `next: __end__` edge or `step_name:outcome` for a routed branch targeting `__end__`. Unlisted terminal edges keep their pause or stop behavior. Use `forge workflow render FILE` instead of tracing a large definition manually. @@ -42,7 +49,7 @@ Dynamic fan-out uses `dynamicRoute: true` and an explicit `maxConcurrency`. Forg The workflow owns topology and flow-level execution choices. `retryBound`, `maxConcurrency`, and join behavior remain valid because they change how the graph advances. -Do not author `kind`, `stationContract`, `stationContractVersion`, `requiredPolicies`, `allowedEffects`, `externalEntry`, `observationPolicy`, `mandatoryPolicies`, or `extensionPoints`. Forge derives node identity, authority, reconciliation, and mandatory governance from the selected state profile. Exceptional commands such as PR rebasing execute through the command-operation boundary and do not appear as workflow steps. Older pinned definitions containing catalog metadata remain readable for compatibility. +Do not author step `kind`, `stationContract`, `stationContractVersion`, `requiredPolicies`, `allowedEffects`, `externalEntry`, `observationPolicy`, `mandatoryPolicies`, or `extensionPoints`. The top-level `kind` is `Workflow` or `Subworkflow`. Forge derives node identity, authority, reconciliation, and mandatory governance from the selected state profile. Exceptional commands such as PR rebasing execute through the command-operation boundary and do not appear as workflow steps. Older pinned definitions containing catalog metadata remain readable for compatibility. Run `forge workflow catalog STATE` when you need to inspect the derived node metadata; do not copy that metadata into the workflow. @@ -72,4 +79,10 @@ forge workflow diff previous.yaml workflow.yaml forge workflow simulate-migration previous.yaml workflow.yaml instances.json ``` -`validate --json` emits canonical storage JSON. `render --format json` emits a compact process manifest. These outputs serve different purposes. +Use `--project-key PROJECT` when a source file includes an active project definition. For a +historical diff or migration check, use the exact expanded artifacts pinned by the instances; +resolving old source files against today's active dependencies can change the graph being compared. + +For a workflow, `validate --json` emits the resolved canonical runtime JSON and +`render --format json` emits a process manifest. For a subworkflow, render JSON emits the +resolved subworkflow definition with its exit targets. diff --git a/README.md b/README.md index 5f270b77..8875e983 100644 --- a/README.md +++ b/README.md @@ -231,7 +231,7 @@ See [Getting Started](https://Forge-sdlc.github.io/forge/getting-started/) for t - [Architecture](https://Forge-sdlc.github.io/forge/architecture/): Versioned workflows, reconciliation, typed stations, durable effects, and execution inspection. - [Declarative Workflows](https://Forge-sdlc.github.io/forge/reference/declarative-workflows/): - Author and govern constrained project workflow definitions. + Author, reuse, and govern constrained project workflow definitions. - [Skills System](https://Forge-sdlc.github.io/forge/skills/): Customize Forge for your team and stack. - [Developer Guide](https://Forge-sdlc.github.io/forge/developer-guide/): Local testing, debugging, Prometheus metrics, Langfuse tracing, and Grafana dashboards. diff --git a/docs/architecture/control-plane.md b/docs/architecture/control-plane.md index 71560cc8..4f8c06b0 100644 --- a/docs/architecture/control-plane.md +++ b/docs/architecture/control-plane.md @@ -49,9 +49,11 @@ are then checked against the active definition's transition policy. ### Pinned declarative definitions Definitions provide the workflow topology: entry, nodes, edges, routed -branches, joins, and flow execution settings. Each workflow instance pins its -name, revision, digest, and canonical definition. Publishing a new definition -changes future selection only; it cannot silently alter an in-flight ticket. +branches, joins, includes, and flow execution settings. Forge expands built-in +and same-project active dependencies into one validated graph before a new run. +Each instance pins the expanded artifact and dependency identities. Activating +a new definition or dependency changes new runs only; it cannot silently alter +an in-flight ticket. The trusted catalog provides authority: node identity, station contracts, allowed effects, mandatory policies, preconditions, and observation behavior. @@ -88,8 +90,8 @@ There are two distinct extension paths. ### Project administrators: declarative configuration Project configuration can select a built-in or published workflow and compose -catalog-registered nodes, routers, transitions, joins, concurrency, and resume -mappings. It can configure repositories, proposal review, skills, and model +catalog-registered nodes, routers, transitions, reusable definitions, joins, +concurrency, and resume mappings. It can configure repositories, proposal review, skills, and model policy. It cannot execute Python or shell code, store credentials, make provider calls, add an effect operation, or relax trusted policies. diff --git a/docs/architecture/reference.md b/docs/architecture/reference.md index ffbc7358..94e28783 100644 --- a/docs/architecture/reference.md +++ b/docs/architecture/reference.md @@ -31,9 +31,10 @@ Run implementation tasks in rootless Podman containers on the Worker host instea ### Golden paths by issue type -Forge ships versioned Feature, Bug, and Task Takeover definitions. They have distinct planning -stages and reuse registered implementation, CI, and review stations. Project definitions may compose -the registered catalog but cannot add arbitrary executable logic. +Forge ships versioned Feature, Bug, and Task Takeover definitions. Their distinct planning +stages share a reusable GitHub PR, CI, and review response subworkflow. Project definitions may +include built-in or same-project active definitions and compose the registered catalog, but +cannot add arbitrary executable logic. New instances pin the expanded graph. ### Human Approval Gates diff --git a/docs/index.md b/docs/index.md index b256532b..f1efbed3 100644 --- a/docs/index.md +++ b/docs/index.md @@ -51,7 +51,7 @@ graph TD - [Operations](operations.md) — Services, execution inspection, recovery, and incident response - [Developer Guide](developer-guide.md) — Full local development reference - [Architecture](architecture/index.md) — How workflow state, reconciliation, stations, and effects fit together -- [Declarative Workflows](reference/declarative-workflows.md) — Compose registered Forge stages safely +- [Declarative Workflows](reference/declarative-workflows.md) — Compose and reuse registered Forge stages safely - [Configuration](reference/config.md) — Repository, model, proposal, and deployment settings - [Skills System](skills/index.md) — Customize Forge for your stack - [Contributing](dev/contributing.md) — How to contribute diff --git a/docs/reference/declarative-workflows.md b/docs/reference/declarative-workflows.md index 4938f8df..b8bedb70 100644 --- a/docs/reference/declarative-workflows.md +++ b/docs/reference/declarative-workflows.md @@ -1,8 +1,9 @@ # Declarative workflows -Forge project administrators can compose the nodes and state profiles shipped with Forge into -project-specific workflows. Definitions are selected by Jira label, validated before compilation, -and compiled into LangGraph graphs at runtime. They cannot import Python or define expressions. +Forge project administrators can compose registered nodes and reusable definitions into +project-specific workflows. Jira labels select active workflows. Forge resolves their includes, +validates the resulting graph, and compiles it with LangGraph. Definitions cannot import Python +or define expressions. This is project-level configuration, not a plugin system. A definition controls flow; trusted Forge code controls authority. To add a provider adapter, a station, a node, or an effect executor, @@ -10,7 +11,9 @@ use the [core-maintainer extension path](../architecture/control-plane.md#core-m ## Author and publish -Create a YAML file locally: +Create a YAML file locally. This small example illustrates the format and passes local +validation; publishing a workflow requires the mandatory gates for its state profile. Start +from the matching full built-in definition for a publishable process. ```yaml apiVersion: forge/v1 @@ -36,11 +39,13 @@ spec: next: prd_approval_gate ``` -Validate and publish it: +Validate the file. Once the complete workflow passes publication validation, publish and +activate its revision as separate decisions: ```bash forge workflow validate workflow.yaml -forge workflow publish MYPROJ workflow.yaml +forge workflow publish MYPROJ complete-workflow.yaml --actor alice --reason "approved process" +forge workflow activate MYPROJ workflow-name 1 --actor alice --reason "start new runs" ``` Before authoring a definition, inspect the supported catalog and before activating a revision, @@ -54,19 +59,20 @@ forge workflow diff previous.yaml workflow.yaml forge workflow simulate-migration previous.yaml workflow.yaml instances.json ``` -Publishing stores canonical JSON in the `forge.workflow.prd-only` Jira project property. Jira -requires the credentials used by the command to have global or project administration permission. -The canonical value must fit Jira's 32,768-byte project-property limit. +Publication stores immutable source JSON and an audit decision in Forge's project-scoped Redis +definition store. Activation sets the pointer used for new runs. Existing Jira project properties +remain a legacy read fallback for workflows that have no governed active definition. A canonical +source or expanded artifact must fit the 32,768-byte definition limit. -Apply `forge:workflow:prd-only` to a ticket to select the workflow. With no such label, Forge uses -its built-in ticket-type routing. Multiple workflow labels, missing definitions, or invalid -definitions block execution instead of silently falling back. +Apply `forge:workflow:workflow-name` to a ticket to select an active workflow. With no such +label, Forge uses its built-in ticket-type routing. Multiple workflow labels, missing definitions, +or invalid definitions block execution instead of silently falling back. ## Format -The checked-in built-in definitions are canonical JSON because that is the exact artifact Forge -pins and stores. They are not intended to be read as raw topology. Render one as Mermaid or as a -compact process manifest instead: +The checked-in built-in definitions are canonical JSON source documents. Forge resolves their +includes before compiling and pinning an instance. Render one as Mermaid or as a compact +process manifest to inspect the expanded topology: ```bash forge workflow render src/forge/workflow/declarative/definitions/feature.json @@ -77,12 +83,87 @@ Authors may use YAML, as in the example above; publishing converts it to canonic format, the fields that describe the process are `spec.entry` and `spec.steps`. Each step declares either a fixed `next` step or a named `route` with possible `branches`. +### Reusable definitions + +Use `kind: Subworkflow` to publish a reusable group of registered steps. Give it an entry, +steps, and named exit targets such as `@exit/review`. A subworkflow can list +`spec.compatibleStates` when the same nodes and routers are valid in several profiles. It is +not selectable by a ticket label. A normal `Workflow` can also be included. Its +`returnFrom` list identifies the terminal edges that return to the caller; all +other `__end__` edges retain their original behavior, including pauses and +blocked termination. + +```yaml +apiVersion: forge/v1 +kind: Subworkflow +metadata: {name: shared_pr, revision: 1} +spec: + state: feature + compatibleStates: [feature, bug, task_takeover] + entry: create_pr + steps: + create_pr: + route: route_after_pr_creation + branches: + escalate_blocked: "@exit/blocked" + teardown_workspace: teardown_workspace + teardown_workspace: + route: route_after_teardown + branches: + human_review_gate: "@exit/review" + setup_workspace: "@exit/setup" +``` + +Include it in a workflow with `spec.includes` and bind every named exit: + +```yaml +spec: + state: feature + entry: generate_prd + includes: + - source: builtin # or project, for an active definition in this Jira project + name: github_pr_review + exits: + blocked: escalate_blocked + review: human_review_gate + setup: setup_workspace +``` + +The caller may target included nodes by their canonical names. An included definition may +appear once per caller; node-name collisions, missing exits, dependency cycles, and profile +mismatches are rejected. For a full workflow include, use `returnTo: caller_step` and +`returnFrom: [step_name]` for a fixed `next: __end__` edge, or +`returnFrom: [step_name:outcome]` for a routed branch to `__end__`. Select each +normal completion edge explicitly; use no `exits`. Includes can be nested. All +expanded steps must satisfy the ordinary workflow +catalog, policy, size, and graph checks. +Subworkflow definitions may set `state`, `compatibleStates`, `entry`, `steps`, and +`includes`. Flow-level `observationPolicy`, `mandatoryPolicies`, `extensionPoints`, +and `resume.fromRevisions` are rejected on subworkflows because inclusion does not +apply those fields to the caller. A project dependency must be published and +activated before another definition can include it. + +Forge ships `github_pr_review` for the shared PR creation, CI, CI fix, and review response +path. Each built-in workflow keeps its own human review gate routing. Project definitions +can include active definitions from the same project. A new run resolves the active +dependency revisions and pins the expanded artifact and dependency identities. Running +instances continue with their pinned graph when a dependency is updated. Forge also stores +the expanded artifact by digest so a checkpoint containing only its revision and digest +can resume after the active dependency changes. + +Use `forge workflow validate FILE --project-key PROJECT` or `render`, `diff`, and +`simulate-migration` with the same option to inspect a definition that includes project +dependencies. Built-in dependencies resolve without a project key. For a historical diff or +migration simulation, use the exact expanded artifacts pinned by the instances; source files +resolved against today's active dependencies may produce a different graph. + Repository users can ask a compatible coding agent to use the generic `.agents/skills/forge-workflow-authoring` skill to create, explain, change, or review a definition. The skill authors YAML and uses Forge's validator, renderer, diff, and migration simulation rather than asking users to edit canonical JSON directly. -- `metadata.name` is lowercase and becomes both the property and label suffix. +- `metadata.name` is lowercase and is the project-scoped publication identity. Workflow names + are also Jira label suffixes; subworkflows are reusable dependencies and cannot be selected. - `metadata.revision` must increase whenever content changes. - `spec.state` is `feature`, `bug`, or `task_takeover` and controls the available node catalog. - Each step name is a canonical, registered Forge node. A step has either `next` or `route` with a @@ -96,8 +177,9 @@ than asking users to edit canonical JSON directly. change how the flow executes. A dynamic router's possible targets are capabilities of its trusted implementation and are derived from the catalog rather than repeated in the workflow. - Graphs may contain a cycle only when it crosses an approved human/CI pause boundary. -- A new instance pins the selected definition's name, revision, digest, and canonical artifact. - Publishing or activating a newer revision does not silently change an active instance. +- A new instance pins the expanded definition's name, revision, digest, dependency identities, + and canonical artifact. Publishing or activating a newer workflow or dependency revision does + not silently change an active instance. To move a pinned instance when a newer revision removes or renames its saved node, add an explicit migration mapping and run compatibility simulation before activation: @@ -110,9 +192,9 @@ spec: old_gate: replacement_gate ``` -State-profile changes, revision rollback, and content changes without a revision increment are -rejected. Published revisions are immutable and retained for pinned instances. Removing an active -pointer prevents new selection but does not mutate an existing checkpoint. +State-profile changes cannot migrate active instances, and changed content needs a new revision. +Published revisions are immutable and retained for pinned instances. A governed rollback may +activate an older compatible revision without changing an existing checkpoint. ## Operational safeguards @@ -157,6 +239,8 @@ Project authors can configure these flow-level choices: - the built-in state profile (`feature`, `bug`, or `task_takeover`); - registered steps, fixed edges, router branches, joins, dynamic fan-out, retry bounds, and concurrency; and +- includes of built-in and same-project active subworkflows or workflows, with explicit exits + or return targets; and - revision/resume mappings for explicitly migratable saved positions. The trusted catalog, not a project definition, owns node kind, station contract, @@ -174,6 +258,5 @@ approval, or make arbitrary network, shell, or Python calls. 5. Diff it against the active revision. 6. When a saved node changes, add `spec.resume.fromRevisions` and run migration simulation against representative active instances. -7. Publish only after reviewing the resulting canonical JSON and migration - result. Existing tickets remain pinned to their prior definition unless a - declared migration applies. +7. Publish and activate separately after reviewing the resolved canonical JSON and migration + result. Existing tickets remain pinned to their prior graph unless explicitly migrated. diff --git a/src/forge/cli.py b/src/forge/cli.py index 4dbdd744..a63ea09b 100644 --- a/src/forge/cli.py +++ b/src/forge/cli.py @@ -20,7 +20,7 @@ def setup_logging(verbose: bool = False) -> None: ) -async def _get_compiled_workflow_for_ticket(ticket_key: str): +async def _get_compiled_workflow_for_ticket(ticket_key: str) -> tuple[Any, Any]: """Helper to get compiled workflow for a ticket (used by CLI commands). Args: @@ -1721,18 +1721,21 @@ def main(argv: list[str] | None = None) -> int: workflow_validate = workflow_subparsers.add_parser("validate", help="Validate a YAML file") workflow_validate.add_argument("file") workflow_validate.add_argument("--json", action="store_true", help="Print canonical JSON") + workflow_validate.add_argument("--project-key", help="Resolve active project dependencies") workflow_render = workflow_subparsers.add_parser( "render", help="Render a validated workflow process manifest" ) workflow_render.add_argument("file") workflow_render.add_argument("--format", choices=("mermaid", "json"), default="mermaid") + workflow_render.add_argument("--project-key", help="Resolve active project dependencies") workflow_diff = workflow_subparsers.add_parser( "diff", help="Report structural and in-flight impact between revisions" ) workflow_diff.add_argument("previous") workflow_diff.add_argument("current") + workflow_diff.add_argument("--project-key", help="Resolve active project dependencies") workflow_simulate = workflow_subparsers.add_parser( "simulate-migration", @@ -1741,6 +1744,7 @@ def main(argv: list[str] | None = None) -> int: workflow_simulate.add_argument("previous") workflow_simulate.add_argument("current") workflow_simulate.add_argument("instances", help="JSON array of active checkpoint snapshots") + workflow_simulate.add_argument("--project-key", help="Resolve active project dependencies") workflow_catalog = workflow_subparsers.add_parser( "catalog", help="Show registered nodes, routers, contracts, and effect authority" diff --git a/src/forge/integrations/jira/client.py b/src/forge/integrations/jira/client.py index 20fd5d5d..0fbbff12 100644 --- a/src/forge/integrations/jira/client.py +++ b/src/forge/integrations/jira/client.py @@ -36,6 +36,7 @@ MAX_RETRIES = 5 INITIAL_BACKOFF_SECONDS = 1.0 MAX_BACKOFF_SECONDS = 60.0 +JIRA_SUMMARY_MAX_LENGTH = 255 _BARE_URL_RE = re.compile(r"https?://[^\s<>]+") _URL_TRAILING_PUNCTUATION = ".,;:!?)]}" @@ -278,7 +279,7 @@ async def create_epic( fields = { "project": {"key": project_key}, - "summary": summary, + "summary": summary[:JIRA_SUMMARY_MAX_LENGTH], "description": adf_content, "issuetype": {"name": "Epic"}, "parent": {"key": parent_key}, @@ -1008,6 +1009,8 @@ async def set_workflow_label( and label != ForgeLabel.FORGE_MANAGED.value and label != "forge:managed:task" and label != "forge:managed:task-takeover" + # YOLO is a workflow setting, not a phase label. + and label != ForgeLabel.YOLO.value # A declarative workflow label identifies the graph definition. It # is not a transient phase label and must survive phase changes. and not label.startswith("forge:workflow:") @@ -1374,8 +1377,11 @@ async def search_issues( "jql": jql, "maxResults": page_size, } - if fields: - params["fields"] = ",".join(fields) + # Jira's enhanced JQL endpoint returns only issue IDs when fields + # is omitted. Always request enough data to populate JiraIssue. + params["fields"] = ",".join( + fields or ["summary", "description", "status", "issuetype", "parent", "labels"] + ) if next_page_token: params["nextPageToken"] = next_page_token diff --git a/src/forge/integrations/source_control/observations.py b/src/forge/integrations/source_control/observations.py index d6d49df7..0dc5f66b 100644 --- a/src/forge/integrations/source_control/observations.py +++ b/src/forge/integrations/source_control/observations.py @@ -51,6 +51,14 @@ def normalized_event_to_observation( resource_type = "check" external_id = f"{event.repo_ref.id}:{event.check.name}" revision = _check_revision(event) + elif event.kind is EventKind.CHECK_UPDATED and event.check_suite_status is not None: + # A check suite is an event about a CI run, not a revision of the PR. + # In particular, a new run on another head SHA must not conflict with + # the previous suite (or with a pull_request event for the same PR). + # The delivery ID is stable for poller redelivery and webhook retries. + resource_type = "check_suite_event" + external_id = f"{event.repo_ref.id}:{event.id}" + revision = event.id elif event.comment: resource_type = "comment" external_id = f"{event.repo_ref.id}:{event.comment.id}" diff --git a/src/forge/workflow/declarative/builtins.py b/src/forge/workflow/declarative/builtins.py index 174f1acf..141c1392 100644 --- a/src/forge/workflow/declarative/builtins.py +++ b/src/forge/workflow/declarative/builtins.py @@ -13,7 +13,7 @@ from forge.models.workflow import TicketType from forge.workflow.declarative.loader import load_workflow_value -from forge.workflow.declarative.models import WorkflowDefinition +from forge.workflow.declarative.models import MAX_STEPS, WorkflowDefinition, WorkflowInclude from forge.workflow.declarative.workflow import DeclarativeWorkflow POLICY = "forge-contracts-v1" @@ -30,6 +30,56 @@ def _load_builtin_definition(name: str) -> WorkflowDefinition: raise RuntimeError(f"missing built-in workflow artifact: {name}") from exc except json.JSONDecodeError as exc: raise RuntimeError(f"invalid built-in workflow artifact: {name}") from exc + if value.get("kind") == "Workflow" and value.get("spec", {}).get("includes"): + # Built-in artifacts are resolved synchronously because their graphs are + # also constructed inside an already-running event loop. + includes = value["spec"].pop("includes") + steps = value["spec"].get("steps") + if not isinstance(steps, dict): + raise ValueError(f"built-in workflow '{name}' requires a steps mapping") + dependencies = [] + for raw_include in includes: + include = WorkflowInclude.model_validate(raw_include) + if include.source != "builtin": + raise RuntimeError("built-in workflows may include only built-in subworkflows") + fragment = _load_builtin_definition(include.name) + if fragment.kind != "Subworkflow": + raise RuntimeError("built-in include must be a subworkflow") + declared_exits = { + target[6:] + for step in fragment.spec.steps.values() + for target in ([step.next] if step.next else step.branches.values()) + if target and target.startswith("@exit/") + } + if ( + set(include.exits) != declared_exits + or include.return_to is not None + or include.return_from + ): + raise ValueError( + f"built-in subworkflow '{include.name}' requires exactly these exits: " + f"{', '.join(sorted(declared_exits))}" + ) + dependencies.append( + f"builtin:{fragment.metadata.name}:{fragment.metadata.revision}:{fragment.digest}" + ) + for node, step in fragment.spec.steps.items(): + if node in steps: + raise RuntimeError(f"duplicate built-in node '{node}'") + imported = step.model_dump(by_alias=True, exclude_none=True) + if imported.get("next", "").startswith("@exit/"): + imported["next"] = include.exits[imported["next"][6:]] + if "branches" in imported: + imported["branches"] = { + outcome: include.exits[target[6:]] + if target.startswith("@exit/") + else target + for outcome, target in imported["branches"].items() + } + steps[node] = imported + if len(steps) > MAX_STEPS: + raise ValueError(f"expanded built-in workflow exceeds {MAX_STEPS} steps") + value["spec"]["resolvedDependencies"] = dependencies definition = load_workflow_value(value) if definition.metadata.name != name: raise RuntimeError( @@ -38,6 +88,13 @@ def _load_builtin_definition(name: str) -> WorkflowDefinition: return definition +def load_builtin_source(name: str) -> WorkflowDefinition: + """Load a checked-in workflow or reusable subworkflow by exact name.""" + if name not in {"feature", "bug", "task_takeover", "github_pr_review"}: + raise ValueError(f"unknown built-in definition '{name}'") + return _load_builtin_definition(name) + + def builtin_feature_definition() -> WorkflowDefinition: """Return the immutable feature golden-path definition.""" return _load_builtin_definition("feature") diff --git a/src/forge/workflow/declarative/cli.py b/src/forge/workflow/declarative/cli.py index 88f054d3..498bbc39 100644 --- a/src/forge/workflow/declarative/cli.py +++ b/src/forge/workflow/declarative/cli.py @@ -9,6 +9,7 @@ import yaml # type: ignore[import-untyped] from forge.workflow.declarative.compiler import DeclarativeWorkflowCompiler +from forge.workflow.declarative.composition import resolve_definition, validate_subworkflow from forge.workflow.declarative.loader import load_workflow_file from forge.workflow.declarative.manifest import ( build_process_manifest, @@ -19,6 +20,14 @@ 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 + ) + + def _print_error(exc: Exception) -> int: print(f"Error: {exc}", file=sys.stderr) return 1 @@ -28,8 +37,15 @@ async def cmd_workflow(args: Any) -> int: action = args.workflow_command if action == "validate": try: - definition = load_workflow_file(args.file) - DeclarativeWorkflowCompiler(definition).validate() + definition = await _load_resolved(args.file, getattr(args, "project_key", None)) + if definition.kind == "Subworkflow": + for state in definition.spec.compatible_states or (definition.spec.state,): + profile_definition = await _load_resolved( + args.file, getattr(args, "project_key", None), state + ) + validate_subworkflow(profile_definition) + else: + DeclarativeWorkflowCompiler(definition).validate() except Exception as exc: return _print_error(exc) print( @@ -42,7 +58,24 @@ async def cmd_workflow(args: Any) -> int: if action == "render": try: - definition = load_workflow_file(args.file) + definition = await _load_resolved(args.file, getattr(args, "project_key", None)) + if definition.kind == "Subworkflow": + validate_subworkflow(definition) + if args.format == "json": + print(json.dumps(definition.canonical_dict(), indent=2)) + else: + lines = ["flowchart TD", f" __start__([start]) --> {definition.spec.entry}"] + for name, step in sorted(definition.spec.steps.items()): + lines.append(f' {name}["{name}"]') + if step.next: + lines.append(f" {name} --> {step.next.replace('@exit/', 'exit_')}") + else: + for outcome, target in sorted(step.branches.items()): + lines.append( + f" {name} -->|{outcome}| {target.replace('@exit/', 'exit_')}" + ) + print("\n".join(lines)) + return 0 manifest = build_process_manifest(definition) except Exception as exc: return _print_error(exc) @@ -54,8 +87,8 @@ async def cmd_workflow(args: Any) -> int: if action == "diff": try: - previous = load_workflow_file(args.previous) - current = load_workflow_file(args.current) + previous = await _load_resolved(args.previous, getattr(args, "project_key", None)) + current = await _load_resolved(args.current, getattr(args, "project_key", None)) impact = compare_process_definitions(previous, current) except Exception as exc: return _print_error(exc) @@ -64,8 +97,8 @@ async def cmd_workflow(args: Any) -> int: if action == "simulate-migration": try: - previous = load_workflow_file(args.previous) - current = load_workflow_file(args.current) + previous = await _load_resolved(args.previous, getattr(args, "project_key", None)) + current = await _load_resolved(args.current, getattr(args, "project_key", None)) with open(args.instances, encoding="utf-8") as source: instances = json.load(source) if not isinstance(instances, list): @@ -157,7 +190,11 @@ async def cmd_workflow(args: Any) -> int: active_definition = await publisher.active(args.name) if active_definition is None: raise ValueError(f"workflow '{args.name}' is not defined for {project_key}") - DeclarativeWorkflowCompiler(active_definition).validate() + if active_definition.kind == "Workflow": + resolved = await resolve_definition(active_definition, publisher.active) + DeclarativeWorkflowCompiler(resolved).validate() + else: + validate_subworkflow(await resolve_definition(active_definition, publisher.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 27d9d011..f274c2bb 100644 --- a/src/forge/workflow/declarative/compiler.py +++ b/src/forge/workflow/declarative/compiler.py @@ -46,6 +46,8 @@ def dynamic_targets(self, step: Any) -> frozenset[str]: def validate(self) -> None: spec = self.definition.spec + if self.definition.kind != "Workflow" or spec.includes: + raise WorkflowValidationError("resolve includes before compiling a workflow") steps = spec.steps if spec.entry not in steps: raise WorkflowValidationError(f"entry node '{spec.entry}' is not declared") diff --git a/src/forge/workflow/declarative/composition.py b/src/forge/workflow/declarative/composition.py new file mode 100644 index 00000000..2b4c2f95 --- /dev/null +++ b/src/forge/workflow/declarative/composition.py @@ -0,0 +1,208 @@ +"""Resolve reusable definitions into the flat, checkpoint-safe process graph.""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable + +from forge.workflow.declarative.models import MAX_STEPS, WorkflowDefinition, WorkflowStep + +ProjectLookup = Callable[[str], Awaitable[WorkflowDefinition | 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 + + if definition.kind != "Subworkflow": + raise ValueError("expected a subworkflow") + unsupported = [ + name + for name, present in ( + ("observationPolicy", definition.spec.observation_policy is not None), + ("mandatoryPolicies", bool(definition.spec.mandatory_policies)), + ("extensionPoints", bool(definition.spec.extension_points)), + ("resume.fromRevisions", bool(definition.spec.resume.from_revisions)), + ) + if present + ] + if unsupported: + raise ValueError( + "subworkflow fields are not applied to the caller: " + ", ".join(unsupported) + ) + if definition.spec.includes: + raise ValueError("publish nested subworkflows after their includes are resolved") + if definition.spec.entry not in definition.spec.steps: + raise ValueError("subworkflow entry is not declared") + if not definition.spec.steps: + raise ValueError("subworkflow requires steps") + 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: + 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}'") + profile.effect_policies[name].resolve(step.allowed_effects) + targets = [step.next] if step.next else step.branches.values() + for target in targets: + if ( + target != "__end__" + and not target.startswith("@exit/") + and target not in definition.spec.steps + ): + raise ValueError(f"subworkflow step '{name}' has undeclared target '{target}'") + if target.startswith("@exit/") and not target[6:]: + raise ValueError("subworkflow exit name is required") + + +async def resolve_definition( + definition: WorkflowDefinition, + project_lookup: ProjectLookup | None = None, + *, + state_profile: str | None = None, +) -> WorkflowDefinition: + """Pin all active dependencies and replace includes with their trusted steps.""" + + async def expand( + source: WorkflowDefinition, chain: tuple[str, ...], profile: str + ) -> tuple[dict[str, WorkflowStep], tuple[str, ...]]: + steps = dict(source.spec.steps) + dependencies: list[str] = [] + seen: set[tuple[str, str]] = set() + included_by: dict[str, str] = {} + for include in source.spec.includes: + key = (include.source, include.name) + if key in seen: + raise ValueError( + f"definition '{include.name}' may be included only once per caller" + ) + seen.add(key) + identity = f"{include.source}:{include.name}" + if identity in chain: + raise ValueError(f"workflow dependency cycle: {' -> '.join((*chain, identity))}") + child: WorkflowDefinition | None + if include.source == "builtin": + from forge.workflow.declarative.builtins import load_builtin_source + + child = load_builtin_source(include.name) + else: + if project_lookup is None: + raise ValueError( + f"project dependency '{include.name}' needs a project resolver" + ) + child = await project_lookup(include.name) + if child is None: + raise ValueError( + f"project dependency '{include.name}' is not active; " + "publish and activate it before including it" + ) + if child.metadata.name != include.name: + raise ValueError(f"dependency '{include.name}' has a mismatched name") + allowed = child.spec.compatible_states or (child.spec.state,) + if profile not in allowed: + raise ValueError(f"dependency '{include.name}' does not support '{profile}'") + child_steps, nested = await expand(child, (*chain, identity), profile) + dependencies.append(f"{identity}:{child.metadata.revision}:{child.digest}") + dependencies.extend(nested) + if child.kind == "Subworkflow": + declared_exits = { + target[6:] + for step in child_steps.values() + for target in ([step.next] if step.next else step.branches.values()) + if target and target.startswith("@exit/") + } + if ( + set(include.exits) != declared_exits + or include.return_to is not None + or include.return_from + ): + raise ValueError( + f"subworkflow '{include.name}' requires exactly these exits: " + f"{', '.join(sorted(declared_exits))}" + ) + child_steps = { + name: _replace_targets( + step, {f"@exit/{port}": target for port, target in include.exits.items()} + ) + for name, step in child_steps.items() + } + else: + if include.exits or include.return_to is None or not include.return_from: + raise ValueError( + f"workflow '{include.name}' requires returnTo and explicit returnFrom edges" + ) + child_steps = _bind_workflow_returns( + child_steps, include.return_from, include.return_to + ) + collision = set(steps) & set(child_steps) + if collision: + node = sorted(collision)[0] + if node in included_by: + raise ValueError( + f"included node '{node}' appears through both " + f"'{included_by[node]}' and '{identity}'" + ) + raise ValueError(f"included node '{node}' collides with caller") + steps.update(child_steps) + included_by.update(dict.fromkeys(child_steps, identity)) + if len(steps) > MAX_STEPS: + raise ValueError(f"expanded workflow exceeds {MAX_STEPS} steps") + return steps, tuple(dependencies) + + if definition.spec.resolved_dependencies: + if definition.spec.includes: + raise ValueError("resolved artifact cannot contain includes") + return definition + steps, dependencies = await expand( + definition, (f"root:{definition.metadata.name}",), state_profile or definition.spec.state + ) + resolved = definition.model_copy( + update={ + "spec": definition.spec.model_copy( + update={ + "steps": steps, + "includes": (), + "resolved_dependencies": dependencies, + } + ) + } + ) + resolved.validate_property_size() + return resolved + + +def _replace_targets(step: WorkflowStep, replacements: dict[str, str]) -> WorkflowStep: + return step.model_copy( + update={ + "next": replacements.get(step.next, step.next) if step.next else None, + "branches": { + outcome: replacements.get(target, target) + for outcome, target in step.branches.items() + }, + } + ) + + +def _bind_workflow_returns( + steps: dict[str, WorkflowStep], selectors: tuple[str, ...], return_to: str +) -> dict[str, WorkflowStep]: + """Bind only the child terminal edges explicitly selected by the caller.""" + if len(selectors) != len(set(selectors)): + raise ValueError("returnFrom contains duplicate edges") + bound = dict(steps) + for selector in selectors: + node, separator, outcome = selector.partition(":") + step = bound.get(node) + if step is None: + raise ValueError(f"returnFrom edge '{selector}' has no declared step") + if separator: + if not outcome or step.branches.get(outcome) != "__end__": + raise ValueError(f"returnFrom edge '{selector}' is not a terminal branch") + bound[node] = step.model_copy( + update={"branches": {**step.branches, outcome: return_to}} + ) + else: + if step.next != "__end__": + raise ValueError(f"returnFrom edge '{selector}' is not a fixed terminal") + bound[node] = step.model_copy(update={"next": return_to}) + return bound diff --git a/src/forge/workflow/declarative/definitions/bug.json b/src/forge/workflow/declarative/definitions/bug.json index 5605cca9..7bf3ee19 100644 --- a/src/forge/workflow/declarative/definitions/bug.json +++ b/src/forge/workflow/declarative/definitions/bug.json @@ -4,10 +4,21 @@ "metadata": { "description": "Forge supported bug-fix golden path", "name": "bug", - "revision": 5 + "revision": 6 }, "spec": { "entry": "triage_check", + "includes": [ + { + "exits": { + "blocked": "escalate_blocked", + "review": "human_review_gate", + "setup": "setup_workspace" + }, + "name": "github_pr_review", + "source": "builtin" + } + ], "resume": { "fromRevisions": {} }, @@ -31,34 +42,6 @@ "dynamicRoute": false, "route": "route_after_answer" }, - "attempt_ci_fix": { - "branches": { - "attempt_ci_fix": "escalate_blocked", - "ci_evaluator": "ci_evaluator", - "escalate_blocked": "escalate_blocked", - "human_review_gate": "human_review_gate" - }, - "dynamicRoute": false, - "retryBound": 5, - "route": "route_current_node" - }, - "ci_evaluator": { - "branches": { - "attempt_ci_fix": "attempt_ci_fix", - "escalate_blocked": "escalate_blocked", - "human_review_gate": "human_review_gate" - }, - "dynamicRoute": false, - "route": "route_ci_evaluation" - }, - "create_pr": { - "branches": { - "escalate_blocked": "escalate_blocked", - "teardown_workspace": "teardown_workspace" - }, - "dynamicRoute": false, - "route": "route_after_pr_creation" - }, "decompose_plan": { "branches": { "__end__": "__end__", @@ -84,32 +67,21 @@ "dynamicRoute": false, "route": "route_human_review_bug" }, - "implement_work": { - "branches": { - "escalate_blocked": "escalate_blocked", - "implement_work": "implement_work", + "implement_work": { + "branches": { + "escalate_blocked": "escalate_blocked", + "implement_work": "implement_work", "local_review": "local_review" }, "dynamicRoute": false, "retryBound": 100, "route": "route_after_implementation" }, - "implement_review": { - "branches": { - "escalate_blocked": "escalate_blocked", - "human_review_gate": "human_review_gate", - "implement_review": "implement_review", - "review_response_gate": "review_response_gate" - }, - "dynamicRoute": false, - "retryBound": 3, - "route": "route_current_node" - }, - "local_review": { + "local_review": { "branches": { "create_pr": "create_pr", "escalate_blocked": "escalate_blocked", - "implement_work": "implement_work", + "implement_work": "implement_work", "local_review": "local_review", "update_documentation": "update_documentation" }, @@ -180,31 +152,14 @@ "dynamicRoute": false, "next": "analyze_bug" }, - "review_response_gate": { + "setup_workspace": { "branches": { - "__end__": "__end__", - "human_review_gate": "human_review_gate", - "implement_review": "implement_review" - }, - "dynamicRoute": false, - "route": "route_review_response" - }, - "setup_workspace": { - "branches": { - "escalate_blocked": "escalate_blocked", - "implement_work": "implement_work" + "escalate_blocked": "escalate_blocked", + "implement_work": "implement_work" }, "dynamicRoute": false, "route": "route_after_workspace_setup" }, - "teardown_workspace": { - "branches": { - "human_review_gate": "human_review_gate", - "setup_workspace": "setup_workspace" - }, - "dynamicRoute": false, - "route": "route_after_teardown" - }, "triage_check": { "branches": { "analyze_bug": "analyze_bug", diff --git a/src/forge/workflow/declarative/definitions/feature.json b/src/forge/workflow/declarative/definitions/feature.json index 89bc2d69..f1c05e2c 100644 --- a/src/forge/workflow/declarative/definitions/feature.json +++ b/src/forge/workflow/declarative/definitions/feature.json @@ -4,10 +4,21 @@ "metadata": { "description": "Forge supported feature golden path", "name": "feature", - "revision": 4 + "revision": 5 }, "spec": { "entry": "generate_prd", + "includes": [ + { + "exits": { + "blocked": "escalate_blocked", + "review": "human_review_gate", + "setup": "setup_workspace" + }, + "name": "github_pr_review", + "source": "builtin" + } + ], "resume": { "fromRevisions": {} }, @@ -33,39 +44,11 @@ "dynamicRoute": false, "route": "route_after_answer" }, - "attempt_ci_fix": { - "branches": { - "attempt_ci_fix": "escalate_blocked", - "ci_evaluator": "ci_evaluator", - "escalate_blocked": "escalate_blocked", - "human_review_gate": "human_review_gate" - }, - "dynamicRoute": false, - "retryBound": 5, - "route": "route_current_node" - }, - "ci_evaluator": { - "branches": { - "attempt_ci_fix": "attempt_ci_fix", - "escalate_blocked": "escalate_blocked", - "human_review_gate": "human_review_gate" - }, - "dynamicRoute": false, - "route": "route_ci_evaluation" - }, "complete_tasks": { "branches": {}, "dynamicRoute": false, "next": "aggregate_epic_status" }, - "create_pr": { - "branches": { - "escalate_blocked": "escalate_blocked", - "teardown_workspace": "teardown_workspace" - }, - "dynamicRoute": false, - "route": "route_after_pr_creation" - }, "decompose_epics": { "branches": { "__end__": "__end__", @@ -113,21 +96,10 @@ "dynamicRoute": false, "route": "route_human_review" }, - "implement_review": { + "implement_work": { "branches": { "escalate_blocked": "escalate_blocked", - "human_review_gate": "human_review_gate", - "implement_review": "implement_review", - "review_response_gate": "review_response_gate" - }, - "dynamicRoute": false, - "retryBound": 3, - "route": "route_current_node" - }, - "implement_work": { - "branches": { - "escalate_blocked": "escalate_blocked", - "implement_work": "implement_work", + "implement_work": "implement_work", "local_review": "local_review" }, "dynamicRoute": false, @@ -156,11 +128,6 @@ "dynamicRoute": false, "route": "route_plan_approval" }, - "provision_epics": { - "branches": {}, - "dynamicRoute": false, - "next": "generate_tasks" - }, "prd_approval_gate": { "branches": { "__end__": "__end__", @@ -171,6 +138,16 @@ "dynamicRoute": false, "route": "route_prd_approval" }, + "provision_epics": { + "branches": {}, + "dynamicRoute": false, + "next": "generate_tasks" + }, + "provision_tasks": { + "branches": {}, + "dynamicRoute": false, + "next": "task_router" + }, "regenerate_all_epics": { "branches": { "__end__": "__end__", @@ -211,19 +188,10 @@ "dynamicRoute": false, "route": "route_after_spec_regeneration" }, - "review_response_gate": { + "setup_workspace": { "branches": { - "__end__": "__end__", - "human_review_gate": "human_review_gate", - "implement_review": "implement_review" - }, - "dynamicRoute": false, - "route": "route_review_response" - }, - "setup_workspace": { - "branches": { - "escalate_blocked": "escalate_blocked", - "implement_work": "implement_work" + "escalate_blocked": "escalate_blocked", + "implement_work": "implement_work" }, "dynamicRoute": false, "route": "route_after_workspace_setup" @@ -242,34 +210,21 @@ "branches": { "__end__": "__end__", "answer_question": "answer_question", + "provision_tasks": "provision_tasks", "regenerate_all_tasks": "regenerate_all_tasks", "regenerate_epic_tasks": "regenerate_epic_tasks", - "provision_tasks": "provision_tasks", "task_router": "task_router", "update_single_task": "update_single_task" }, "dynamicRoute": false, "route": "route_task_approval" }, - "provision_tasks": { - "branches": {}, - "dynamicRoute": false, - "next": "task_router" - }, "task_router": { "branches": {}, "dynamicRoute": true, "maxConcurrency": 16, "route": "route_tasks_parallel" }, - "teardown_workspace": { - "branches": { - "human_review_gate": "human_review_gate", - "setup_workspace": "setup_workspace" - }, - "dynamicRoute": false, - "route": "route_after_teardown" - }, "update_documentation": { "branches": {}, "dynamicRoute": false, diff --git a/src/forge/workflow/declarative/definitions/github_pr_review.json b/src/forge/workflow/declarative/definitions/github_pr_review.json new file mode 100644 index 00000000..7bf3b357 --- /dev/null +++ b/src/forge/workflow/declarative/definitions/github_pr_review.json @@ -0,0 +1,76 @@ +{ + "apiVersion": "forge/v1", + "kind": "Subworkflow", + "metadata": { + "description": "Shared pull request, CI, and review response path", + "name": "github_pr_review", + "revision": 1 + }, + "spec": { + "compatibleStates": [ + "feature", + "bug", + "task_takeover" + ], + "entry": "create_pr", + "state": "feature", + "steps": { + "attempt_ci_fix": { + "branches": { + "attempt_ci_fix": "@exit/blocked", + "ci_evaluator": "ci_evaluator", + "escalate_blocked": "@exit/blocked", + "human_review_gate": "@exit/review" + }, + "dynamicRoute": false, + "retryBound": 5, + "route": "route_current_node" + }, + "ci_evaluator": { + "branches": { + "attempt_ci_fix": "attempt_ci_fix", + "escalate_blocked": "@exit/blocked", + "human_review_gate": "@exit/review" + }, + "dynamicRoute": false, + "route": "route_ci_evaluation" + }, + "create_pr": { + "branches": { + "escalate_blocked": "@exit/blocked", + "teardown_workspace": "teardown_workspace" + }, + "dynamicRoute": false, + "route": "route_after_pr_creation" + }, + "implement_review": { + "branches": { + "escalate_blocked": "@exit/blocked", + "human_review_gate": "@exit/review", + "implement_review": "implement_review", + "review_response_gate": "review_response_gate" + }, + "dynamicRoute": false, + "retryBound": 3, + "route": "route_current_node" + }, + "review_response_gate": { + "branches": { + "__end__": "__end__", + "human_review_gate": "@exit/review", + "implement_review": "implement_review" + }, + "dynamicRoute": false, + "route": "route_review_response" + }, + "teardown_workspace": { + "branches": { + "human_review_gate": "@exit/review", + "setup_workspace": "@exit/setup" + }, + "dynamicRoute": false, + "route": "route_after_teardown" + } + } + } +} diff --git a/src/forge/workflow/declarative/definitions/task_takeover.json b/src/forge/workflow/declarative/definitions/task_takeover.json index f43b96fc..689297ee 100644 --- a/src/forge/workflow/declarative/definitions/task_takeover.json +++ b/src/forge/workflow/declarative/definitions/task_takeover.json @@ -4,10 +4,21 @@ "metadata": { "description": "Forge supported task-takeover golden path", "name": "task_takeover", - "revision": 5 + "revision": 6 }, "spec": { "entry": "triage_check", + "includes": [ + { + "exits": { + "blocked": "escalate_blocked", + "review": "human_review_gate", + "setup": "setup_workspace" + }, + "name": "github_pr_review", + "source": "builtin" + } + ], "resume": { "fromRevisions": {} }, @@ -20,54 +31,16 @@ "dynamicRoute": false, "route": "route_after_answer" }, - "attempt_ci_fix": { - "branches": { - "attempt_ci_fix": "escalate_blocked", - "ci_evaluator": "ci_evaluator", - "escalate_blocked": "escalate_blocked", - "human_review_gate": "human_review_gate" - }, - "dynamicRoute": false, - "retryBound": 5, - "route": "route_current_node" - }, - "ci_evaluator": { - "branches": { - "attempt_ci_fix": "attempt_ci_fix", - "escalate_blocked": "escalate_blocked", - "human_review_gate": "human_review_gate" - }, - "dynamicRoute": false, - "route": "route_ci_evaluation" - }, "complete_task_takeover": { "branches": {}, "dynamicRoute": false, "next": "__end__" }, - "create_pr": { - "branches": { - "escalate_blocked": "escalate_blocked", - "teardown_workspace": "teardown_workspace" - }, - "dynamicRoute": false, - "route": "route_after_pr_creation" - }, "escalate_blocked": { "branches": {}, "dynamicRoute": false, "next": "__end__" }, - "implement_work": { - "branches": { - "escalate_blocked": "escalate_blocked", - "implement_work": "implement_work", - "run_qualitative_review": "run_qualitative_review" - }, - "dynamicRoute": false, - "retryBound": 100, - "route": "route_after_execution" - }, "generate_plan": { "branches": { "escalate_blocked": "escalate_blocked", @@ -89,41 +62,31 @@ "dynamicRoute": false, "route": "route_human_review_task_takeover" }, - "implement_review": { + "implement_work": { "branches": { "escalate_blocked": "escalate_blocked", - "human_review_gate": "human_review_gate", - "implement_review": "implement_review", - "review_response_gate": "review_response_gate" - }, - "dynamicRoute": false, - "retryBound": 3, - "route": "route_current_node" - }, - "review_response_gate": { - "branches": { - "__end__": "__end__", - "human_review_gate": "human_review_gate", - "implement_review": "implement_review" + "implement_work": "implement_work", + "run_qualitative_review": "run_qualitative_review" }, "dynamicRoute": false, - "route": "route_review_response" + "retryBound": 100, + "route": "route_after_execution" }, - "run_qualitative_review": { + "run_qualitative_review": { "branches": { "create_pr": "create_pr", "escalate_blocked": "escalate_blocked", - "implement_work": "implement_work", + "implement_work": "implement_work", "run_qualitative_review": "run_qualitative_review" }, "dynamicRoute": false, "retryBound": 3, "route": "route_after_qualitative_review" }, - "setup_workspace": { - "branches": { - "escalate_blocked": "escalate_blocked", - "implement_work": "implement_work" + "setup_workspace": { + "branches": { + "escalate_blocked": "escalate_blocked", + "implement_work": "implement_work" }, "dynamicRoute": false, "route": "route_after_workspace_setup" @@ -138,14 +101,6 @@ "dynamicRoute": false, "route": "route_task_plan_approval" }, - "teardown_workspace": { - "branches": { - "human_review_gate": "human_review_gate", - "setup_workspace": "setup_workspace" - }, - "dynamicRoute": false, - "route": "route_after_teardown" - }, "triage_check": { "branches": { "escalate_blocked": "escalate_blocked", diff --git a/src/forge/workflow/declarative/models.py b/src/forge/workflow/declarative/models.py index 468f9fb1..1c51f9ba 100644 --- a/src/forge/workflow/declarative/models.py +++ b/src/forge/workflow/declarative/models.py @@ -104,10 +104,37 @@ class WorkflowResume(StrictModel): ) +class WorkflowInclude(StrictModel): + source: Literal["builtin", "project"] + name: str + exits: dict[str, str] = Field(default_factory=dict) + return_to: str | None = Field(default=None, alias="returnTo") + # Full workflow includes must identify which terminal edges return to the + # caller. Other __end__ edges may represent waits or blocked termination. + return_from: tuple[str, ...] = Field( + default=(), alias="returnFrom", exclude_if=lambda value: not value + ) + + @field_validator("name") + @classmethod + def validate_name(cls, value: str) -> str: + if not WORKFLOW_NAME_RE.fullmatch(value): + raise ValueError("invalid included definition name") + return value + + class WorkflowSpec(StrictModel): state: Literal["feature", "bug", "task_takeover"] entry: str - steps: dict[str, WorkflowStep] + steps: dict[str, WorkflowStep] = Field(default_factory=dict) + includes: tuple[WorkflowInclude, ...] = Field(default=(), exclude_if=lambda value: not value) + resolved_dependencies: tuple[str, ...] = Field( + default=(), alias="resolvedDependencies", exclude_if=lambda value: not value + ) + # Subworkflows can be checked against more than one trusted state profile. + compatible_states: tuple[Literal["feature", "bug", "task_takeover"], ...] = Field( + default=(), alias="compatibleStates", exclude_if=lambda value: not value + ) # The provider-neutral policy used to apply external observations to this # workflow instance. Policies are versioned, allowlisted runtime # capabilities; arbitrary import paths are deliberately not supported. @@ -134,8 +161,6 @@ def validate_entry(cls, value: str) -> str: @field_validator("steps") @classmethod def validate_steps(cls, value: dict[str, WorkflowStep]) -> dict[str, WorkflowStep]: - if not value: - raise ValueError("at least one step is required") if len(value) > MAX_STEPS: raise ValueError(f"a workflow may have at most {MAX_STEPS} steps") invalid = [name for name in value if not NODE_NAME_RE.fullmatch(name)] @@ -143,13 +168,29 @@ def validate_steps(cls, value: dict[str, WorkflowStep]) -> dict[str, WorkflowSte raise ValueError(f"invalid canonical node name: {invalid[0]}") return value + @model_validator(mode="after") + def validate_body(self) -> WorkflowSpec: + if not self.steps and not self.includes: + raise ValueError("at least one step or include is required") + return self + class WorkflowDefinition(StrictModel): api_version: Literal["forge/v1"] = Field(alias="apiVersion") - kind: Literal["Workflow"] + kind: Literal["Workflow", "Subworkflow"] metadata: WorkflowMetadata spec: WorkflowSpec + @model_validator(mode="after") + def validate_kind(self) -> WorkflowDefinition: + if self.kind == "Workflow" and self.spec.compatible_states: + raise ValueError("compatibleStates is only valid for subworkflows") + if self.kind == "Subworkflow" and self.spec.state not in ( + self.spec.compatible_states or (self.spec.state,) + ): + raise ValueError("subworkflow state must be compatible") + return self + def canonical_dict(self) -> dict[str, Any]: return self.model_dump(by_alias=True, exclude_none=True, mode="json") diff --git a/src/forge/workflow/declarative/publication.py b/src/forge/workflow/declarative/publication.py index b2c3d32f..1af74737 100644 --- a/src/forge/workflow/declarative/publication.py +++ b/src/forge/workflow/declarative/publication.py @@ -10,10 +10,12 @@ from forge.domain import DomainModel from forge.orchestrator.checkpointer import get_redis_client from forge.workflow.declarative.compiler import DeclarativeWorkflowCompiler +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 _DEFINITION_PREFIX = "forge:process:def:" +_EXPANDED_PREFIX = "forge:process:expanded:" _ACTIVE_PREFIX = "forge:process:active:" _DECISIONS_PREFIX = "forge:process:decisions:" _LATEST_PREFIX = "forge:process:latest:" @@ -73,6 +75,30 @@ def _compatible_impact(impact: ProcessChangeImpact, *, rollback: bool = False) - return impact.compatible_for_in_flight +def _affected_consumers( + active: dict[str, WorkflowDefinition], name: str +) -> tuple[WorkflowDefinition, ...]: + """Find active workflows that depend on a candidate, including nested includes.""" + affected = {name} + changed = True + while changed: + changed = False + for consumer_name, definition in active.items(): + if consumer_name in affected: + continue + if any( + include.source == "project" and include.name in affected + for include in definition.spec.includes + ): + affected.add(consumer_name) + changed = True + return tuple( + definition + for consumer_name, definition in active.items() + if consumer_name != name and consumer_name in affected and definition.kind == "Workflow" + ) + + class DefinitionPublisher: """Project-scoped immutable definition store and rollout decision log.""" @@ -93,7 +119,12 @@ async def publish( """Validate and persist an immutable artifact, without activating it.""" if activate: raise ValueError("publication and activation are separate decisions; use activate()") - self._validate(definition) + await self._validate(definition) + if any( + previous.kind != definition.kind + for previous in await self.history(definition.metadata.name) + ): + raise ValueError("definition kind cannot change across revisions of the same name") decision = self._decision(definition, actor=actor, reason=reason, action="publish") result = await (await self._client()).eval( _PUBLISH_SCRIPT, @@ -170,8 +201,11 @@ async def _set_active( raise ValueError(f"published workflow '{name}' revision {revision} is unavailable") if target.metadata.name != name: raise ValueError("published workflow name does not match activation key") - self._validate(target) + await self._validate(target) + await self._validate_consumers(name, target) previous = await self.active(name) + if previous is not None and previous.kind != target.kind: + raise ValueError("definition kind cannot change across revisions of the same name") if previous is not None and expected_active_digest is None: raise ValueError( "expected_active_digest is required when replacing an active definition" @@ -180,7 +214,12 @@ async def _set_active( previous is None or previous.digest != expected_active_digest ): raise ValueError("active definition changed concurrently") - impact = compare_process_definitions(previous, target) if previous else None + 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), + ) if impact is not None and not _compatible_impact(impact, rollback=action == "rollback"): raise ValueError("definition is incompatible with active workflow instances") decision = self._decision( @@ -206,6 +245,19 @@ async def get(self, name: str, revision: int) -> WorkflowDefinition | None: value = await (await self._client()).get(self._definition_key(name, revision)) return WorkflowDefinition.model_validate_json(value) if value else None + async def remember_expanded(self, definition: WorkflowDefinition) -> None: + """Keep the exact graph used by a run for identity-only checkpoint recovery.""" + key = self._expanded_key( + definition.metadata.name, definition.metadata.revision, definition.digest + ) + await (await self._client()).set(key, definition.canonical_json(), nx=True) + + async def get_expanded( + self, name: str, revision: int, digest: str + ) -> WorkflowDefinition | None: + value = await (await self._client()).get(self._expanded_key(name, revision, digest)) + return WorkflowDefinition.model_validate_json(value) if value else None + async def active(self, name: str) -> WorkflowDefinition | None: pointer = await (await self._client()).get(self._active_key(name)) if not pointer: @@ -250,9 +302,29 @@ async def list_workflows(self) -> tuple[str, ...]: break return tuple(sorted(names)) - def _validate(self, definition: WorkflowDefinition) -> None: + async def _validate(self, definition: WorkflowDefinition) -> None: definition.validate_property_size() - DeclarativeWorkflowCompiler(definition).validate_for_publication() + 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) + ) + else: + resolved = await resolve_definition(definition, self.active) + DeclarativeWorkflowCompiler(resolved).validate_for_publication() + + async def _validate_consumers(self, name: str, candidate: WorkflowDefinition) -> None: + async def lookup(dependency: str) -> WorkflowDefinition | None: + return candidate if dependency == name else await self.active(dependency) + + active = { + consumer_name: consumer + for consumer_name in await self.list_workflows() + if (consumer := await self.active(consumer_name)) is not None + } + for consumer in _affected_consumers(active, name): + resolved = await resolve_definition(consumer, lookup) + DeclarativeWorkflowCompiler(resolved).validate_for_publication() def _decision( self, @@ -297,6 +369,9 @@ def _prefix(self, prefix: str, name: str) -> str: def _definition_key(self, name: str, revision: int | str) -> str: return f"{self._prefix(_DEFINITION_PREFIX, name)}:{revision}" + def _expanded_key(self, name: str, revision: int, digest: str) -> str: + return f"{self._prefix(_EXPANDED_PREFIX, name)}:{revision}:{digest}" + def _latest_key(self, name: str) -> str: return self._prefix(_LATEST_PREFIX, name) @@ -319,6 +394,7 @@ def __init__(self, project_key: str = "DEFAULT") -> None: raise ValueError("project_key is required for governed publication") self.project_key = project_key.upper() self._definitions: dict[tuple[str, int], WorkflowDefinition] = {} + self._expanded: dict[tuple[str, int, str], WorkflowDefinition] = {} self._active: dict[str, WorkflowDefinition] = {} self._decisions: dict[str, list[PublicationDecision]] = {} @@ -327,12 +403,14 @@ async def publish( ) -> PublicationDecision: if activate: raise ValueError("publication and activation are separate decisions; use activate()") - self._validate(definition) + await self._validate(definition) key = (definition.metadata.name, definition.metadata.revision) existing = self._definitions.get(key) if existing is not None and existing.digest != definition.digest: raise ValueError("published revision is immutable and has different content") published = await self.history(definition.metadata.name) + if any(previous.kind != definition.kind for previous in published): + raise ValueError("definition kind cannot change across revisions of the same name") if any( item.digest != definition.digest and item.metadata.revision >= definition.metadata.revision @@ -400,14 +478,23 @@ async def _set_active( raise ValueError(f"published workflow '{name}' revision {revision} is unavailable") if target.metadata.name != name: raise ValueError("published workflow name does not match activation key") + await self._validate(target) + await self._validate_consumers(name, target) current = self._active.get(name) + if current is not None and current.kind != target.kind: + raise ValueError("definition kind cannot change across revisions of the same name") if current is not None and expected_active_digest is None: raise ValueError( "expected_active_digest is required when replacing an active definition" ) if expected_active_digest and (current is None or current.digest != expected_active_digest): raise ValueError("active definition changed concurrently") - impact = compare_process_definitions(current, target) if current else None + 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), + ) if impact and not _compatible_impact(impact, rollback=action == "rollback"): raise ValueError("definition is incompatible with active workflow instances") self._active[name] = target @@ -420,6 +507,15 @@ async def _set_active( async def get(self, name: str, revision: int) -> WorkflowDefinition | None: return self._definitions.get((name, revision)) + async def remember_expanded(self, definition: WorkflowDefinition) -> None: + key = (definition.metadata.name, definition.metadata.revision, definition.digest) + self._expanded.setdefault(key, definition) + + async def get_expanded( + self, name: str, revision: int, digest: str + ) -> WorkflowDefinition | None: + return self._expanded.get((name, revision, digest)) + async def active(self, name: str) -> WorkflowDefinition | None: return self._active.get(name) @@ -437,9 +533,29 @@ async def history(self, name: str) -> tuple[WorkflowDefinition, ...]: async def list_workflows(self) -> tuple[str, ...]: return tuple(sorted({name for name, _ in self._definitions})) - def _validate(self, definition: WorkflowDefinition) -> None: + async def _validate(self, definition: WorkflowDefinition) -> None: definition.validate_property_size() - DeclarativeWorkflowCompiler(definition).validate_for_publication() + 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) + ) + else: + resolved = await resolve_definition(definition, self.active) + DeclarativeWorkflowCompiler(resolved).validate_for_publication() + + async def _validate_consumers(self, name: str, candidate: WorkflowDefinition) -> None: + async def lookup(dependency: str) -> WorkflowDefinition | None: + return candidate if dependency == name else await self.active(dependency) + + active = { + consumer_name: consumer + for consumer_name in await self.list_workflows() + if (consumer := await self.active(consumer_name)) is not None + } + for consumer in _affected_consumers(active, name): + resolved = await resolve_definition(consumer, lookup) + DeclarativeWorkflowCompiler(resolved).validate_for_publication() def _decision( self, diff --git a/src/forge/workflow/declarative/resolver.py b/src/forge/workflow/declarative/resolver.py index 712e1284..1d5f866b 100644 --- a/src/forge/workflow/declarative/resolver.py +++ b/src/forge/workflow/declarative/resolver.py @@ -4,11 +4,13 @@ from typing import Any, Protocol +from forge.workflow.declarative.composition import resolve_definition from forge.workflow.declarative.loader import load_workflow_value from forge.workflow.declarative.models import ( WORKFLOW_LABEL_PREFIX, WORKFLOW_NAME_RE, WORKFLOW_PROPERTY_PREFIX, + WorkflowDefinition, ) from forge.workflow.declarative.workflow import DeclarativeWorkflow @@ -18,6 +20,10 @@ async def get(self, name: str, revision: int) -> Any | None: ... async def active(self, name: str) -> Any | None: ... + async def remember_expanded(self, definition: Any) -> None: ... + + async def get_expanded(self, name: str, revision: int, digest: str) -> Any | None: ... + class ProjectPropertyReader(Protocol): async def get_project_property(self, project_key: str, property_key: str) -> Any | None: ... @@ -54,7 +60,7 @@ async def load_project_workflow( A checkpoint's canonical definition is preferred because it is the durable source of truth for an in-flight instance. If only identity metadata was - persisted, ``definition_reader`` must provide the exact published revision; + persisted, ``definition_reader`` must provide the exact expanded artifact; this function deliberately never falls back to Jira's active property for a pinned checkpoint. """ @@ -71,7 +77,12 @@ async def load_project_workflow( raise ValueError( f"published workflow '{workflow_name}' revision {pinned_revision} is unavailable" ) - value = await definition_reader.get(workflow_name, int(pinned_revision)) + value = await definition_reader.get_expanded( + workflow_name, int(pinned_revision), pinned_digest + ) + if value is None: + # Older, non-composed runs only have a published source artifact. + value = await definition_reader.get(workflow_name, int(pinned_revision)) if value is None: raise ValueError( f"published workflow '{workflow_name}' revision {pinned_revision} is unavailable" @@ -100,6 +111,38 @@ 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: + loaded_dependencies: dict[str, str] = {} + dependency_cache: dict[str, WorkflowDefinition | None] = {} + + async def dependency(name: str) -> WorkflowDefinition | None: + if definition_reader is None: + return None + if name in dependency_cache: + return dependency_cache[name] + value = await definition_reader.active(name) + resolved: WorkflowDefinition | None = ( + value + if isinstance(value, WorkflowDefinition) + else load_workflow_value(value) + if value + else None + ) + if resolved is not None: + loaded_dependencies[name] = resolved.digest + dependency_cache[name] = resolved + return resolved + + definition = await resolve_definition(definition, dependency) + for name, digest in loaded_dependencies.items(): + current = await definition_reader.active(name) if definition_reader else None + if current is None: + raise ValueError(f"project dependency '{name}' changed during resolution") + current = current if hasattr(current, "digest") else load_workflow_value(current) + if current.digest != digest: + raise ValueError(f"project dependency '{name}' changed during resolution") + if definition_reader is not None: + await definition_reader.remember_expanded(definition) if definition.metadata.name != workflow_name: raise ValueError( f"workflow property name '{workflow_name}' does not match metadata name " diff --git a/src/forge/workflow/gates/plan_approval.py b/src/forge/workflow/gates/plan_approval.py index c0c9dbee..f32f8fa9 100644 --- a/src/forge/workflow/gates/plan_approval.py +++ b/src/forge/workflow/gates/plan_approval.py @@ -126,26 +126,79 @@ async def provision_epics(state: WorkflowState) -> WorkflowState: async def provision_epics_from_draft(state: WorkflowState, jira: "JiraClient") -> list[str]: """Materialize the approved workflow-state draft as Jira Epics.""" + from forge.integrations.jira.client import JIRA_SUMMARY_MAX_LENGTH from forge.models.draft import ForgeDecompositionDraft from forge.models.workflow import ForgeLabel ticket_key = state["ticket_key"] existing = await jira.search_issues( - f'labels = "forge:parent:{ticket_key}" AND issuetype = Epic' + f'labels = "forge:parent:{ticket_key}" AND issuetype = Epic', + fields=["summary", "labels"], + max_results=None, ) - if existing: - return [issue.key for issue in existing] raw = state.get("plan_draft") if not raw: raise ValueError(f"Approved plan_draft not found for {ticket_key}") draft = ForgeDecompositionDraft.model_validate(raw) if isinstance(raw, dict) else raw project_key = (await jira.get_issue(ticket_key)).project_key + active_items = [item for item in draft.items if not item.excluded] + marker_prefix = f"forge:epic-item:{ticket_key}:" + claimed: set[str] = set() + matches: dict[int, str] = {} + + # Stable per-item labels survive title edits and make a resumed provision + # independent of Jira search ordering. Old Epics without markers are + # matched by summary once and then upgraded to the stable label. + for item in active_items: + marker = f"{marker_prefix}{item.id}" + candidates = [issue for issue in existing if marker in getattr(issue, "labels", [])] + if len(candidates) > 1: + raise ValueError(f"Epic marker {marker} matches multiple Jira issues") + if candidates: + matches[item.id] = candidates[0].key + claimed.add(candidates[0].key) + + for item in active_items: + if item.id in matches: + continue + summary = item.summary[:JIRA_SUMMARY_MAX_LENGTH] + candidates = [ + issue + for issue in existing + if issue.key not in claimed + and issue.summary == summary + and not any(label.startswith(marker_prefix) for label in getattr(issue, "labels", [])) + ] + if len(candidates) > 1: + raise ValueError(f"Epic summary {summary!r} matches multiple unmarked Jira issues") + if candidates: + matches[item.id] = candidates[0].key + claimed.add(candidates[0].key) + await jira.add_labels(candidates[0].key, [f"{marker_prefix}{item.id}"]) + + unclaimed = [ + issue.key + for issue in existing + if issue.key not in claimed + and not any(label.startswith(marker_prefix) for label in getattr(issue, "labels", [])) + ] + if unclaimed: + raise ValueError( + f"Cannot safely reconcile unmarked Epics for {ticket_key}: {', '.join(unclaimed)}" + ) + epic_keys: list[str] = [] - for item in draft.items: - if item.excluded: + for item in active_items: + existing_key = matches.get(item.id) + if existing_key: + epic_keys.append(existing_key) continue - labels = [ForgeLabel.FORGE_MANAGED.value, f"forge:parent:{ticket_key}"] + labels = [ + ForgeLabel.FORGE_MANAGED.value, + f"forge:parent:{ticket_key}", + f"{marker_prefix}{item.id}", + ] if item.repo and "/" in item.repo: labels.append(f"repo:{item.repo}") epic_keys.append( diff --git a/src/forge/workflow/nodes/task_generation.py b/src/forge/workflow/nodes/task_generation.py index de6c28cd..339505e0 100644 --- a/src/forge/workflow/nodes/task_generation.py +++ b/src/forge/workflow/nodes/task_generation.py @@ -49,7 +49,7 @@ async def generate_tasks(state: WorkflowState) -> WorkflowState: # Revision-3 workflows created before draft provisioning became an # explicit declarative node route approval directly to generate_tasks. - if not epic_keys and state.get("plan_draft"): + if (not epic_keys or any(not key for key in epic_keys)) and state.get("plan_draft"): from forge.workflow.gates.plan_approval import provision_epics_from_draft jira = JiraClient() diff --git a/tests/integration/workflow/test_definition_publication.py b/tests/integration/workflow/test_definition_publication.py index a71ed91b..90166549 100644 --- a/tests/integration/workflow/test_definition_publication.py +++ b/tests/integration/workflow/test_definition_publication.py @@ -58,3 +58,17 @@ async def test_redis_publication_activation_and_cas(redis_client) -> None: first.metadata.revision, second.metadata.revision, ] + + +@pytest.mark.asyncio +async def test_expanded_artifact_survives_publisher_recreation(redis_client) -> None: + first = builtin_feature_definition() + publisher = DefinitionPublisher("PROJ", redis_client=redis_client) + await publisher.remember_expanded(first) + + reopened = DefinitionPublisher("PROJ", redis_client=redis_client) + recovered = await reopened.get_expanded( + first.metadata.name, first.metadata.revision, first.digest + ) + assert recovered is not None + assert recovered.digest == first.digest diff --git a/tests/unit/integrations/jira/test_client.py b/tests/unit/integrations/jira/test_client.py index 5c6cbb5d..df575fd7 100644 --- a/tests/unit/integrations/jira/test_client.py +++ b/tests/unit/integrations/jira/test_client.py @@ -303,6 +303,7 @@ async def test_set_workflow_label_preserves_declarative_workflow_identity(self, return_value=[ "forge:managed", "forge:workflow:planning-smoke", + "forge:yolo", "forge:prd-pending", ] ) @@ -318,9 +319,24 @@ async def test_set_workflow_label_preserves_declarative_workflow_identity(self, operations = mock_http.put.call_args.kwargs["json"]["update"]["labels"] assert {"remove": "forge:workflow:planning-smoke"} not in operations + assert {"remove": "forge:yolo"} not in operations assert {"remove": "forge:prd-pending"} in operations +class TestJiraClientCreateEpic: + @pytest.mark.asyncio + async def test_generated_summary_is_limited_to_jira_maximum(self): + jira = JiraClient() + response = MagicMock() + response.json.return_value = {"key": "AISOS-123"} + http = AsyncMock() + http.post.return_value = response + with patch.object(jira, "_get_client", return_value=http): + key = await jira.create_epic("AISOS", "x" * 294, "Description", "AISOS-1") + assert key == "AISOS-123" + assert len(http.post.call_args.kwargs["json"]["fields"]["summary"]) == 255 + + class TestJiraClientArchiveIssue: """Tests for archive_issue method.""" @@ -1092,6 +1108,18 @@ async def test_malformed_response(self, jira_client): class TestJiraClientSearchIssues: + @pytest.mark.asyncio + async def test_requests_fields_when_caller_uses_defaults(self, jira_client): + response = MagicMock() + response.json.return_value = {"issues": [], "isLast": True} + http = AsyncMock() + http.get.return_value = response + + with patch.object(jira_client, "_get_client", return_value=http): + await jira_client.search_issues('project = "PROJ"') + + assert "summary" in http.get.call_args.kwargs["params"]["fields"] + @pytest.mark.asyncio async def test_uses_enhanced_jql_search_and_token_pagination(self, jira_client): first = MagicMock() diff --git a/tests/unit/integrations/source_control/test_observations.py b/tests/unit/integrations/source_control/test_observations.py index 4ee82f6c..7de21c71 100644 --- a/tests/unit/integrations/source_control/test_observations.py +++ b/tests/unit/integrations/source_control/test_observations.py @@ -8,6 +8,7 @@ ChangeRequest, ChangeRequestIdentity, ChangeRequestState, + CheckStatus, EventKind, NormalizedEvent, Provider, @@ -127,4 +128,31 @@ async def test_merged_change_request_is_terminal_and_independent_of_head_sha() - ledger = InMemoryObservationLedger() assert (await ledger.record(updated_observation)).disposition is ObservationDisposition.ACCEPTED assert (await ledger.record(merged_observation)).disposition is ObservationDisposition.ACCEPTED - assert (await ledger.record(redelivery_observation)).disposition is ObservationDisposition.DUPLICATE + assert ( + await ledger.record(redelivery_observation) + ).disposition is ObservationDisposition.DUPLICATE + + +@pytest.mark.asyncio +async def test_check_suite_events_for_successive_ci_runs_are_independent() -> None: + first = _event() + first.id = "poller-check_suite-failed" + first.kind = EventKind.CHECK_UPDATED + first.check_suite_status = CheckStatus.COMPLETED + + second = _event() + second.id = "poller-check_suite-passed" + second.kind = EventKind.CHECK_UPDATED + second.check_suite_status = CheckStatus.COMPLETED + assert second.change_request is not None + second.change_request.head_sha = "def456" + + first_observation = normalized_event_to_observation(first) + second_observation = normalized_event_to_observation(second) + assert first_observation.resource.resource_type == "check_suite_event" + assert first_observation.resource != second_observation.resource + + ledger = InMemoryObservationLedger() + assert (await ledger.record(first_observation)).disposition is ObservationDisposition.ACCEPTED + assert (await ledger.record(second_observation)).disposition is ObservationDisposition.ACCEPTED + assert (await ledger.record(second_observation)).disposition is ObservationDisposition.DUPLICATE diff --git a/tests/unit/workflow/gates/test_plan_draft_state.py b/tests/unit/workflow/gates/test_plan_draft_state.py index fd13d80f..73ed4e47 100644 --- a/tests/unit/workflow/gates/test_plan_draft_state.py +++ b/tests/unit/workflow/gates/test_plan_draft_state.py @@ -1,6 +1,7 @@ """Regression coverage for state-backed epic draft provisioning.""" from datetime import UTC, datetime +from types import SimpleNamespace from unittest.mock import AsyncMock import pytest @@ -38,5 +39,109 @@ async def test_provision_epics_uses_checkpointed_draft_without_attachment_lifecy assert epic_keys == ["AISOS-2"] jira.create_epic.assert_awaited_once() + assert "forge:epic-item:AISOS-1:1" in jira.create_epic.call_args.kwargs["labels"] + assert jira.search_issues.call_args.kwargs["max_results"] is None jira.add_attachment.assert_not_awaited() jira.delete_attachments_by_name.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_provision_epics_resumes_after_one_epic_was_created() -> None: + first_summary = "A" * 294 + draft = ForgeDecompositionDraft( + parent_key="AISOS-1", + phase="epics", + items=[ + DraftItem( + id=1, + summary=first_summary, + description="First", + repo="forge-sdlc/forge", + acceptance_criteria=[], + ), + DraftItem( + id=2, + summary="Second", + description="Second", + repo="forge-sdlc/forge", + acceptance_criteria=[], + ), + ], + version=1, + created_at=datetime.now(UTC), + updated_at=datetime.now(UTC), + ) + jira = AsyncMock() + jira.search_issues.return_value = [SimpleNamespace(key="AISOS-2", summary=first_summary[:255])] + jira.get_issue.return_value.project_key = "AISOS" + jira.create_epic.return_value = "AISOS-3" + + epic_keys = await provision_epics_from_draft( + {"ticket_key": "AISOS-1", "plan_draft": draft}, jira + ) + + assert epic_keys == ["AISOS-2", "AISOS-3"] + jira.create_epic.assert_awaited_once() + jira.add_labels.assert_awaited_once_with("AISOS-2", ["forge:epic-item:AISOS-1:1"]) + assert jira.create_epic.call_args.kwargs["summary"] == "Second" + + +@pytest.mark.asyncio +async def test_provision_epics_reuses_marked_epic_after_title_change() -> None: + draft = ForgeDecompositionDraft( + parent_key="AISOS-1", + phase="epics", + items=[ + DraftItem( + id=1, + summary="Updated title", + description="Updated description", + repo="forge-sdlc/forge", + acceptance_criteria=[], + ) + ], + version=2, + created_at=datetime.now(UTC), + updated_at=datetime.now(UTC), + ) + jira = AsyncMock() + jira.search_issues.return_value = [ + SimpleNamespace( + key="AISOS-2", + summary="Old title", + labels=["forge:parent:AISOS-1", "forge:epic-item:AISOS-1:1"], + ) + ] + jira.get_issue.return_value.project_key = "AISOS" + + assert await provision_epics_from_draft( + {"ticket_key": "AISOS-1", "plan_draft": draft}, jira + ) == ["AISOS-2"] + jira.create_epic.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_provision_epics_stops_on_unmatched_legacy_epic() -> None: + draft = ForgeDecompositionDraft( + parent_key="AISOS-1", + phase="epics", + items=[ + DraftItem( + id=1, + summary="Updated title", + description="Updated description", + repo="forge-sdlc/forge", + acceptance_criteria=[], + ) + ], + version=2, + created_at=datetime.now(UTC), + updated_at=datetime.now(UTC), + ) + jira = AsyncMock() + jira.search_issues.return_value = [SimpleNamespace(key="AISOS-2", summary="Old title")] + jira.get_issue.return_value.project_key = "AISOS" + + with pytest.raises(ValueError, match="Cannot safely reconcile unmarked Epics"): + await provision_epics_from_draft({"ticket_key": "AISOS-1", "plan_draft": draft}, jira) + jira.create_epic.assert_not_awaited() diff --git a/tests/unit/workflow/test_builtin_definition_artifacts.py b/tests/unit/workflow/test_builtin_definition_artifacts.py index e71deb1c..b3968039 100644 --- a/tests/unit/workflow/test_builtin_definition_artifacts.py +++ b/tests/unit/workflow/test_builtin_definition_artifacts.py @@ -8,6 +8,7 @@ import pytest from forge.workflow.declarative.builtins import ( + _load_builtin_definition, builtin_bug_definition, builtin_feature_definition, builtin_task_takeover_definition, @@ -24,9 +25,9 @@ # A changed digest is an intentional process revision and must update the # checked-in artifact and this snapshot together. _DIGESTS = { - "feature": "7764c3ba6a9ede67f9b4b2636c9718085aa07f24a50cb4054b6068fe18ae9841", - "bug": "c78b72f68d8c10bb58a0e019395ee2dff3b184928f25fb4d740b01e4612dc7d1", - "task_takeover": "63690df2b210effda77e00b79d39a3f7be62ffb727b67cc657bf58b47a787b37", + "feature": "5f653ba647ffbb9cbeda762b363e95eacbc17ed99b82a5d1176ac6fa9c9a8a29", + "bug": "95d5afd82d727e9eaa1f75f40635536d32ef1b7ef0af84fd9cca2a29683b1d07", + "task_takeover": "06049d7a4fee8789254b897ece896b004b54a2ae04f2b27a1f173c86129053fb", } @@ -38,7 +39,10 @@ def test_artifact_round_trip_preserves_canonical_definition(name: str) -> None: definition = load_workflow_value(artifact) assert definition.canonical_dict() == artifact - assert _DEFINITIONS[name]().canonical_dict() == artifact + expanded = _DEFINITIONS[name]() + assert not expanded.spec.includes + assert expanded.spec.resolved_dependencies + assert set(expanded.spec.steps) > set(definition.spec.steps) @pytest.mark.parametrize("name", tuple(_DEFINITIONS)) @@ -53,3 +57,36 @@ def test_default_compiler_consumes_checked_in_artifact(name: str) -> None: compiler = DeclarativeWorkflowCompiler(definition) compiler.validate() assert compiler.build_graph() is not None + + +@pytest.mark.parametrize("damage", ["missing_exit", "missing_steps"]) +def test_builtin_include_rejects_malformed_artifact( + damage: str, monkeypatch: pytest.MonkeyPatch +) -> None: + from forge.workflow.declarative import builtins + + real_files = resources.files + artifact = json.loads( + real_files("forge.workflow.declarative.definitions") + .joinpath("feature.json") + .read_text(encoding="utf-8") + ) + if damage == "missing_exit": + del artifact["spec"]["includes"][0]["exits"]["review"] + else: + del artifact["spec"]["steps"] + + class Resource: + def read_text(self, *, encoding: str) -> str: + assert encoding == "utf-8" + return json.dumps(artifact) + + class Root: + def joinpath(self, name: str): + if name == "feature.json": + return Resource() + return real_files("forge.workflow.declarative.definitions").joinpath(name) + + monkeypatch.setattr(builtins.resources, "files", lambda _: Root()) + with pytest.raises(ValueError, match="requires exactly these exits|requires a steps mapping"): + _load_builtin_definition("feature") diff --git a/tests/unit/workflow/test_composition.py b/tests/unit/workflow/test_composition.py new file mode 100644 index 00000000..89f77ca2 --- /dev/null +++ b/tests/unit/workflow/test_composition.py @@ -0,0 +1,533 @@ +"""Reusable process definitions preserve a flat, immutable runtime graph.""" + +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest + +from forge.workflow.declarative.builtins import builtin_feature_definition, load_builtin_source +from forge.workflow.declarative.compiler import DeclarativeWorkflowCompiler +from forge.workflow.declarative.composition import resolve_definition, validate_subworkflow +from forge.workflow.declarative.loader import load_workflow_value +from forge.workflow.declarative.models import WorkflowInclude +from forge.workflow.declarative.publication import ( + InMemoryDefinitionPublisher, + _affected_consumers, +) +from forge.workflow.declarative.resolver import load_project_workflow + + +def _project_consumer(name: str = "shared_feature"): + original = builtin_feature_definition() + fragment = load_builtin_source("github_pr_review") + spec = original.spec.model_copy( + update={ + "steps": { + key: value + for key, value in original.spec.steps.items() + if key not in fragment.spec.steps + }, + "includes": ( + WorkflowInclude( + source="project", + name="github_pr_review", + exits={ + "blocked": "escalate_blocked", + "review": "human_review_gate", + "setup": "setup_workspace", + }, + ), + ), + "resolved_dependencies": (), + } + ) + return original.model_copy( + update={ + "metadata": original.metadata.model_copy(update={"name": name, "revision": 1}), + "spec": spec, + } + ) + + +@pytest.mark.asyncio +async def test_project_dependency_update_affects_new_runs_and_preserves_pin() -> None: + publisher = InMemoryDefinitionPublisher("PROJ") + fragment = load_builtin_source("github_pr_review") + validate_subworkflow(fragment) + await publisher.publish(fragment, actor="admin", reason="shared path") + await publisher.activate(fragment, actor="admin", reason="enable shared path") + consumer = _project_consumer() + await publisher.publish(consumer, actor="admin", reason="use shared path") + await publisher.activate(consumer, actor="admin", reason="enable process") + + first = await load_project_workflow( + None, "PROJ", consumer.metadata.name, definition_reader=publisher + ) + DeclarativeWorkflowCompiler(first.definition).validate_for_publication() + assert first.definition.spec.steps["create_pr"] == fragment.spec.steps["create_pr"].model_copy( + update={ + "branches": { + "escalate_blocked": "escalate_blocked", + "teardown_workspace": "teardown_workspace", + } + } + ) + + updated_steps = dict(fragment.spec.steps) + updated_steps["attempt_ci_fix"] = updated_steps["attempt_ci_fix"].model_copy( + update={"retry_bound": 4} + ) + second_fragment = fragment.model_copy( + update={ + "metadata": fragment.metadata.model_copy(update={"revision": 2}), + "spec": fragment.spec.model_copy(update={"steps": updated_steps}), + } + ) + await publisher.publish(second_fragment, actor="admin", reason="new retry bound") + await publisher.activate( + second_fragment, actor="admin", reason="roll out", expected_active_digest=fragment.digest + ) + second = await load_project_workflow( + None, "PROJ", consumer.metadata.name, definition_reader=publisher + ) + assert second.definition.digest != first.definition.digest + assert second.definition.spec.steps["attempt_ci_fix"].retry_bound == 4 + + pinned = await load_project_workflow( + None, + "PROJ", + consumer.metadata.name, + pinned_revision=first.definition.metadata.revision, + pinned_digest=first.definition.digest, + pinned_definition=first.definition.canonical_dict(), + definition_reader=publisher, + ) + assert pinned.definition.digest == first.definition.digest + + identity_only = await load_project_workflow( + None, + "PROJ", + consumer.metadata.name, + pinned_revision=first.definition.metadata.revision, + pinned_digest=first.definition.digest, + definition_reader=publisher, + ) + assert identity_only.definition.digest == first.definition.digest + assert identity_only.definition.spec.steps["attempt_ci_fix"].retry_bound == 5 + + with pytest.raises(ValueError, match="digest does not match checkpoint"): + await load_project_workflow( + None, + "PROJ", + consumer.metadata.name, + pinned_revision=first.definition.metadata.revision, + pinned_digest="missing-expanded-artifact", + definition_reader=publisher, + ) + + +@pytest.mark.asyncio +async def test_full_workflow_normal_completion_returns_to_caller() -> None: + child = load_builtin_source("feature") + child = child.model_copy( + update={ + "metadata": child.metadata.model_copy(update={"name": "child"}), + "spec": child.spec.model_copy( + update={ + "entry": "generate_prd", + "steps": { + "generate_prd": child.spec.steps["generate_prd"].model_copy( + update={"next": "__end__", "route": None, "branches": {}} + ) + }, + "resolved_dependencies": (), + } + ), + } + ) + parent = _project_consumer("outer") + parent = parent.model_copy( + update={ + "spec": parent.spec.model_copy( + update={ + "entry": "generate_spec", + "steps": { + "generate_spec": child.spec.steps["generate_prd"].model_copy( + update={"next": "generate_prd"} + ), + "answer_question": child.spec.steps["generate_prd"].model_copy( + update={"next": "__end__"} + ), + }, + "includes": ( + WorkflowInclude( + source="project", + name="child", + returnTo="answer_question", + returnFrom=("generate_prd",), + ), + ), + } + ) + } + ) + + async def lookup(name: str): + return child if name == "child" else None + + resolved = await resolve_definition(parent, lookup) + assert resolved.spec.steps["generate_prd"].next == "answer_question" + DeclarativeWorkflowCompiler(resolved).validate() + + +@pytest.mark.asyncio +async def test_full_workflow_include_preserves_wait_and_blocked_terminals() -> None: + child = load_workflow_value( + { + "apiVersion": "forge/v1", + "kind": "Workflow", + "metadata": {"name": "child", "revision": 1}, + "spec": { + "state": "feature", + "entry": "generate_spec", + "steps": { + "generate_spec": { + "route": "route_current_node", + "branches": { + "done": "__end__", + "wait": "__end__", + "blocked": "escalate_blocked", + }, + }, + "escalate_blocked": {"next": "__end__"}, + }, + }, + } + ) + parent = load_workflow_value( + { + "apiVersion": "forge/v1", + "kind": "Workflow", + "metadata": {"name": "parent", "revision": 1}, + "spec": { + "state": "feature", + "entry": "answer_question", + "steps": {"answer_question": {"next": "generate_spec"}}, + "includes": [ + { + "source": "project", + "name": "child", + "returnTo": "answer_question", + "returnFrom": ["generate_spec:done"], + } + ], + }, + } + ) + + async def lookup(name: str): + return child if name == "child" else None + + resolved = await resolve_definition(parent, lookup) + assert resolved.spec.steps["generate_spec"].branches["done"] == "answer_question" + assert resolved.spec.steps["generate_spec"].branches["wait"] == "__end__" + assert resolved.spec.steps["escalate_blocked"].next == "__end__" + + +@pytest.mark.asyncio +async def test_missing_exit_and_dependency_cycles_are_rejected() -> None: + consumer = _project_consumer() + fragment = load_builtin_source("github_pr_review") + bad_include = consumer.spec.includes[0].model_copy( + update={"exits": {"review": "human_review_gate"}} + ) + bad = consumer.model_copy( + update={"spec": consumer.spec.model_copy(update={"includes": (bad_include,)})} + ) + + async def lookup(name: str): + return fragment if name == "github_pr_review" else None + + with pytest.raises(ValueError, match="requires exactly these exits"): + await resolve_definition(bad, lookup) + + recursive = consumer.model_copy( + update={ + "spec": consumer.spec.model_copy( + update={ + "includes": ( + WorkflowInclude( + source="project", name=consumer.metadata.name, returnTo="__end__" + ), + ) + } + ) + } + ) + + async def self_lookup(name: str): + return recursive if name == recursive.metadata.name else None + + with pytest.raises(ValueError, match="dependency cycle"): + await resolve_definition(recursive, self_lookup) + + +@pytest.mark.asyncio +async def test_dependency_activation_rejects_a_graph_that_breaks_active_consumer() -> None: + publisher = InMemoryDefinitionPublisher("PROJ") + fragment = load_builtin_source("github_pr_review") + await publisher.publish(fragment, actor="admin", reason="shared path") + await publisher.activate(fragment, actor="admin", reason="enable shared path") + consumer = _project_consumer() + await publisher.publish(consumer, actor="admin", reason="use shared path") + await publisher.activate(consumer, actor="admin", reason="enable process") + + broken_steps = dict(fragment.spec.steps) + ci_branches = dict(broken_steps["ci_evaluator"].branches) + ci_branches["escalate_blocked"] = "@exit/urgent" + broken_steps["ci_evaluator"] = broken_steps["ci_evaluator"].model_copy( + update={"branches": ci_branches} + ) + broken = fragment.model_copy( + update={ + "metadata": fragment.metadata.model_copy(update={"revision": 2}), + "spec": fragment.spec.model_copy(update={"steps": broken_steps}), + } + ) + await publisher.publish(broken, actor="admin", reason="new exit") + with pytest.raises(ValueError, match="requires exactly these exits"): + await publisher.activate( + broken, actor="admin", reason="bad rollout", expected_active_digest=fragment.digest + ) + assert (await publisher.active("github_pr_review")).digest == fragment.digest + + +@pytest.mark.asyncio +async def test_dependency_activation_ignores_unrelated_invalid_workflow() -> None: + publisher = InMemoryDefinitionPublisher("PROJ") + fragment = load_builtin_source("github_pr_review") + await publisher.publish(fragment, actor="admin", reason="shared path") + await publisher.activate(fragment, actor="admin", reason="enable shared path") + + unrelated = _project_consumer("unrelated") + missing = unrelated.spec.includes[0].model_copy(update={"name": "missing_dependency"}) + unrelated = unrelated.model_copy( + update={"spec": unrelated.spec.model_copy(update={"includes": (missing,)})} + ) + publisher._active["unrelated"] = unrelated + + updated = fragment.model_copy( + update={"metadata": fragment.metadata.model_copy(update={"revision": 2})} + ) + await publisher.publish(updated, actor="admin", reason="new revision") + await publisher.activate( + updated, actor="admin", reason="roll out", expected_active_digest=fragment.digest + ) + assert (await publisher.active("github_pr_review")).digest == updated.digest + + +def test_dependency_impact_follows_nested_project_includes() -> None: + fragment = load_builtin_source("github_pr_review") + middle = fragment.model_copy( + update={ + "metadata": fragment.metadata.model_copy(update={"name": "middle"}), + "spec": fragment.spec.model_copy( + update={ + "includes": (WorkflowInclude(source="project", name="github_pr_review"),), + } + ), + } + ) + outer = _project_consumer("outer") + outer_include = outer.spec.includes[0].model_copy(update={"name": "middle"}) + outer = outer.model_copy( + update={"spec": outer.spec.model_copy(update={"includes": (outer_include,)})} + ) + affected = _affected_consumers( + {"github_pr_review": fragment, "middle": middle, "outer": outer}, "github_pr_review" + ) + assert affected == (outer,) + + +@pytest.mark.parametrize( + "field,value", + [ + ("observation_policy", "default"), + ("mandatory_policies", ("require-review",)), + ("extension_points", ("after-review",)), + ], +) +def test_subworkflow_rejects_caller_fields_that_composition_ignores(field, value) -> None: + fragment = load_builtin_source("github_pr_review") + invalid = fragment.model_copy(update={"spec": fragment.spec.model_copy(update={field: value})}) + with pytest.raises(ValueError, match="not applied to the caller"): + validate_subworkflow(invalid) + + +@pytest.mark.asyncio +async def test_definition_kind_cannot_change_across_revisions() -> None: + publisher = InMemoryDefinitionPublisher("PROJ") + fragment = load_builtin_source("github_pr_review") + publisher._definitions[(fragment.metadata.name, 1)] = fragment + publisher._active[fragment.metadata.name] = fragment + replacement = fragment.model_copy( + update={ + "kind": "Workflow", + "metadata": fragment.metadata.model_copy(update={"revision": 2}), + } + ) + publisher._validate = AsyncMock() + + with pytest.raises(ValueError, match="kind cannot change"): + await publisher.publish(replacement, actor="admin", reason="replace kind") + + publisher._definitions[(fragment.metadata.name, 2)] = replacement + with pytest.raises(ValueError, match="kind cannot change"): + await publisher.activate( + replacement, + actor="admin", + reason="replace kind", + expected_active_digest=fragment.digest, + ) + + +@pytest.mark.asyncio +async def test_collision_and_state_profile_mismatch_are_rejected() -> None: + fragment = load_builtin_source("github_pr_review") + consumer = _project_consumer() + + async def lookup(name: str): + return fragment if name == fragment.metadata.name else None + + colliding_steps = dict(consumer.spec.steps) + colliding_steps["create_pr"] = fragment.spec.steps["create_pr"] + colliding = consumer.model_copy( + update={"spec": consumer.spec.model_copy(update={"steps": colliding_steps})} + ) + with pytest.raises(ValueError, match="collides"): + await resolve_definition(colliding, lookup) + + incompatible = fragment.model_copy( + update={"spec": fragment.spec.model_copy(update={"compatible_states": ()})} + ) + + async def incompatible_lookup(name: str): + return incompatible if name == fragment.metadata.name else None + + bug_caller = consumer.model_copy( + update={"spec": consumer.spec.model_copy(update={"state": "bug"})} + ) + with pytest.raises(ValueError, match="does not support 'bug'"): + await resolve_definition(bug_caller, incompatible_lookup) + + +@pytest.mark.asyncio +async def test_nested_subworkflows_bind_exits_through_parent() -> None: + def definition(kind: str, name: str, entry: str, steps: dict, includes: list | None = None): + return load_workflow_value( + { + "apiVersion": "forge/v1", + "kind": kind, + "metadata": {"name": name, "revision": 1}, + "spec": { + "state": "feature", + "entry": entry, + "steps": steps, + "includes": includes or [], + }, + } + ) + + inner = definition( + "Subworkflow", "inner", "generate_spec", {"generate_spec": {"next": "@exit/done"}} + ) + middle = definition( + "Subworkflow", + "middle", + "generate_prd", + {"generate_prd": {"next": "generate_spec"}}, + [{"source": "project", "name": "inner", "exits": {"done": "@exit/done"}}], + ) + outer = definition( + "Workflow", + "outer", + "generate_prd", + {"answer_question": {"next": "__end__"}}, + [{"source": "project", "name": "middle", "exits": {"done": "answer_question"}}], + ) + + async def lookup(name: str): + return {"inner": inner, "middle": middle}.get(name) + + resolved = await resolve_definition(outer, lookup) + assert resolved.spec.steps["generate_spec"].next == "answer_question" + DeclarativeWorkflowCompiler(resolved).validate() + + +@pytest.mark.asyncio +async def test_diamond_dependency_identifies_duplicate_include_paths() -> None: + def workflow(name: str, step: str, includes: list | None = None): + return load_workflow_value( + { + "apiVersion": "forge/v1", + "kind": "Workflow", + "metadata": {"name": name, "revision": 1}, + "spec": { + "state": "feature", + "entry": step, + "steps": {step: {"next": "__end__"}}, + "includes": includes or [], + }, + } + ) + + shared = workflow("shared", "generate_spec") + left = workflow( + "left", + "generate_prd", + [ + { + "source": "project", + "name": "shared", + "returnTo": "generate_prd", + "returnFrom": ["generate_spec"], + } + ], + ) + right = workflow( + "right", + "answer_question", + [ + { + "source": "project", + "name": "shared", + "returnTo": "answer_question", + "returnFrom": ["generate_spec"], + } + ], + ) + root = workflow( + "root", + "triage_bug", + [ + { + "source": "project", + "name": "left", + "returnTo": "triage_bug", + "returnFrom": ["generate_prd"], + }, + { + "source": "project", + "name": "right", + "returnTo": "triage_bug", + "returnFrom": ["answer_question"], + }, + ], + ) + + async def lookup(name: str): + return {"shared": shared, "left": left, "right": right}.get(name) + + with pytest.raises(ValueError, match="appears through both 'project:left' and 'project:right'"): + await resolve_definition(root, lookup)