From ce0747b5db77dfb6620db6c2a75d13afbaf5a2d6 Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Thu, 27 Aug 2026 15:39:51 +0300 Subject: [PATCH 1/4] feat: establish versioned workflow domain contracts --- docs/architecture/option-b-decoupling-plan.md | 2 + .../phase-1-domain-contracts-plan.md | 235 ++++++++++++++++++ src/forge/domain/__init__.py | 39 +++ src/forge/domain/commands.py | 31 +++ src/forge/domain/effects.py | 39 +++ src/forge/domain/identity.py | 35 +++ src/forge/domain/observations.py | 30 +++ src/forge/domain/schema.py | 23 ++ src/forge/domain/stations.py | 56 +++++ tests/unit/domain/test_architecture.py | 31 +++ tests/unit/domain/test_contracts.py | 119 +++++++++ 11 files changed, 640 insertions(+) create mode 100644 docs/architecture/phase-1-domain-contracts-plan.md create mode 100644 src/forge/domain/__init__.py create mode 100644 src/forge/domain/commands.py create mode 100644 src/forge/domain/effects.py create mode 100644 src/forge/domain/identity.py create mode 100644 src/forge/domain/observations.py create mode 100644 src/forge/domain/schema.py create mode 100644 src/forge/domain/stations.py create mode 100644 tests/unit/domain/test_architecture.py create mode 100644 tests/unit/domain/test_contracts.py diff --git a/docs/architecture/option-b-decoupling-plan.md b/docs/architecture/option-b-decoupling-plan.md index d2cfc5e2..b2b34b1d 100644 --- a/docs/architecture/option-b-decoupling-plan.md +++ b/docs/architecture/option-b-decoupling-plan.md @@ -111,6 +111,8 @@ coupling. ### Phase 1 — Establish versioned domain contracts +**Detailed plan:** [Phase 1 domain contracts plan](phase-1-domain-contracts-plan.md). + Introduce small, Forge-owned contracts independent of LangGraph and providers: - `Observation`: source, external identity, resource identity, observed revision/time, diff --git a/docs/architecture/phase-1-domain-contracts-plan.md b/docs/architecture/phase-1-domain-contracts-plan.md new file mode 100644 index 00000000..6943b09f --- /dev/null +++ b/docs/architecture/phase-1-domain-contracts-plan.md @@ -0,0 +1,235 @@ +# Phase 1 implementation plan: versioned domain contracts + +**Status:** Proposed + +**Depends on:** Phase 0 baseline and the stacked integration of `dev`, PR 317, and +PR 318 + +**Goal:** Introduce Forge-owned, versioned runtime contracts and prove them on one +real station without changing graph behavior, checkpoint compatibility, or external +effects. + +## Outcome + +At the end of Phase 1, Forge will have a provider- and LangGraph-independent contract +layer for observations, workflow commands, station invocations, station outcomes, and +effect intent/results. The existing graphs and `BaseState` remain operational, but one +station will receive only a typed projection of the state and return a validated outcome +which a reducer applies to the checkpoint. + +Phase 1 establishes boundaries; it does not yet extract event interpretation from the +worker, execute effects through a journal, or convert every workflow node into a station. + +## Design decisions + +### Package boundary + +Add a new `forge.domain` package. It may depend on Python and the chosen schema-validation +library, but not on LangGraph, Redis, Jira, GitHub, provider adapters, the worker, or +workflow-specific state types. + +Proposed layout: + +```text +src/forge/domain/ + identity.py # workflow, invocation, resource, and idempotency identities + observations.py # Observation and normalized facts + commands.py # WorkflowCommand and command categories + stations.py # StationRequest, StationOutcome, status, and typed payload protocol + effects.py # EffectCommand, EffectResult, operation and result categories + schema.py # schema-version validation and JSON-safe serialization helpers + +src/forge/workflow/stations/ + implementation_input.py # first provider-independent station + +src/forge/workflow/projections/ + implementation_input.py # checkpoint/provider facts -> station request + +src/forge/workflow/reducers/ + implementation_input.py # validated station outcome -> checkpoint update +``` + +Contract versions are explicit data, not inferred from the installed Forge version. +Version 1 contracts use strict validation, reject unknown status values, and serialize to +JSON-safe primitives. Payload types are scoped to their station rather than accepting +`BaseState` or an arbitrary dictionary. + +### Compatibility approach + +- Keep `BaseState` as the LangGraph checkpoint schema in Phase 1. +- Keep existing graph node names and routes. +- Wrap the first station with a projector and reducer behind the existing node/function + entry point. +- Preserve existing checkpoint fields and legacy planning adapters. +- Convert `NormalizedEvent` into an `Observation`; do not remove or fork the existing + source-control transport format yet. +- Define effect intent contracts, but continue current inline effects until Phase 3. + +This permits rollback to the old implementation without migrating persisted checkpoints. + +## Delivery stages + +### 1.1 — Contract kernel + +Implement the five contract families and their stable identities: + +- `Observation`: schema version, source (`webhook`, `poller`, or internal), observation + identity, external resource identity and revision, observed/received time, normalized + facts, and correlation metadata. +- `WorkflowCommand`: command identity, workflow target, `start`, `resume`, `approve`, + `reject`, `retry`, `cancel`, or `synchronize`, evidence references, and requested time. +- `StationRequest[T]`: workflow/definition identity, invocation identity, contract name + and version, attempt, deadline/policy context, artifact references, and typed input. +- `StationOutcome[T]`: `succeeded`, `blocked`, `waiting`, `retryable_failure`, or + `terminal_failure`, typed output, requested effects, and structured reason/error. +- `EffectCommand` / `EffectResult`: stable idempotency key, provider-neutral operation, + logical target, expected precondition, payload, and durable result category. + +Add round-trip, malformed-input, forward-version rejection, equality, and stable-identity +tests. Add an architecture test preventing `forge.domain` from importing execution or +provider packages. + +**Why it matters:** all later extraction work shares one vocabulary and one compatibility +policy. Invalid station results fail at the boundary instead of becoming corrupt graph +state. + +### 1.2 — Observation compatibility adapter + +Add a lossless adapter from the existing source-control `NormalizedEvent` to +`Observation`. Define stable observation identity from provider event ID plus resource +revision when available, preserve raw payload only as referenced compatibility evidence, +and record whether the source is webhook or poller. + +Do not change queue consumers or worker routing in this stage. Add fixtures proving that +serialization is deterministic and repeated conversion produces the same identity. + +**Why it matters:** the poller and webhooks can converge on a common Forge-owned envelope +without coupling the domain layer to GitHub or prematurely rewriting ingress. + +### 1.3 — Projection and reducer boundary + +Create reusable interfaces for: + +```text +checkpoint + normalized facts -> StationRequest +StationOutcome + checkpoint -> validated state update +``` + +Reducers must allowlist fields, validate the station name and invocation identity, retain +audit metadata, and reject outputs from a different contract version or workflow run. +They return a partial LangGraph update; they cannot mutate the input state. + +Add contract tests for missing inputs, stale invocation IDs, malformed outputs, and +attempted writes outside the allowlist. + +**Why it matters:** this is the actual coupling break. A station no longer reads or +returns the complete feature, bug, or task state merely because LangGraph stores it. + +### 1.4 — First station: implementation-input resolution + +Refactor `resolve_implementation_input` into the first contract-backed station because it +already provides a shared domain operation for feature, bug, and task-takeover flows. + +The scoped input contains only repository identity, candidate work units, planning +artifact references/content required for selection, completion markers, and normalized +work-item snapshots. The projector performs compatibility reads from legacy checkpoint +fields and obtains external work-item facts through the existing narrow reader. The +station imports neither `JiraIssue`, `JiraClient`, `BaseState`, nor workflow-specific +state. Its typed output contains the selected work unit, ordered context artifacts, +instructions, and optional summary. The reducer alone creates the existing `artifacts`, +`work_units`, `current_work_unit_id`, and `work_resolution` checkpoint updates. + +Keep the current `resolve_implementation_input(state, jira)` entry point as a compatibility +facade during Phase 1. Run the existing feature/bug/task tests against both the legacy +behavior fixture and the contract-backed implementation. + +**Why it matters:** it proves that one meaningful operation can be run locally from a +fixture, shared by several graphs, and evolved independently of their full state schemas. + +### 1.5 — Conformance, rollout, and measurement + +Add a station conformance suite covering version negotiation, JSON serialization, +determinism for identical requests, outcome validation, and reducer field ownership. +Expose a small local runner that accepts a serialized `StationRequest` fixture and emits +a serialized `StationOutcome` without starting LangGraph, Redis, or provider clients. + +Regenerate the Phase 0 architecture report and record: + +- complete-state fields formerly read by implementation-input resolution; +- fields present in its new request and writable by its reducer; +- prohibited dependencies removed; +- checkpoint and golden-path test equivalence. + +Initially enable the facade for all flows because it preserves the public call shape. If +behavior differs, retain a temporary compatibility switch for one release and compare +outcomes in tests; do not dual-execute external effects. + +**Why it matters:** the phase ends with an independently testable station and measurable +coupling reduction, not only unused model classes. + +## PR sequence + +Keep Phase 1 reviewable as four stacked PRs on top of PR 318 (or its eventual merged +successor): + +1. **Contract kernel and architecture rule** — Stage 1.1. +2. **Observation adapter** — Stage 1.2. +3. **Projection/reducer framework** — Stage 1.3. +4. **Implementation-input station and conformance runner** — Stages 1.4 and 1.5. + +Each PR must pass the full current gate. Contract additions must not require checkpoint +migration, and the final PR must run feature, bug, and task-takeover characterization +tests. + +## Implications + +### Benefits + +- LangGraph remains authoritative for process position while station code becomes + portable and narrowly scoped. +- Provider replacement becomes less invasive because provider models stop at projection + and adapter boundaries. +- Station contracts become versionable, locally runnable, and suitable for declarative + graph validation in Phase 5. +- Typed outcomes provide the basis for durable effects, retries, and operator read models. + +### Costs and risks + +- During migration, Forge carries both broad checkpoint state and narrow station models, + adding adapters and some duplication. +- Contract versioning creates an ongoing compatibility obligation; versions cannot be + changed casually once checkpoints or queued requests reference them. +- A generic `facts: dict` or payload escape hatch could recreate the current coupling. + Its contents must therefore be typed per observation/station and architecture-tested. +- Moving Jira reads out of the station makes projection code temporarily more complex. +- Effect commands are declarative only in this phase; inline side effects remain a known + crash/replay risk until Phase 3. + +## Non-goals + +- Replacing `BaseState` or migrating existing checkpoints. +- Rewriting `OrchestratorWorker` event dispatch (Phase 2). +- Executing or persisting effect commands (Phase 3). +- Migrating all nodes into stations (Phase 4). +- Changing graph topology, routing, or approval policy. +- Moving polling into Forge or making poller cursor state authoritative. + +## Exit criteria + +Phase 1 is complete only when: + +1. All five contract families are versioned, strict, JSON round-trippable, and free of + LangGraph/provider dependencies. +2. `NormalizedEvent` has a deterministic, tested conversion to `Observation` without + breaking the existing queue format. +3. Implementation-input resolution consumes a typed `StationRequest` and produces a + validated `StationOutcome` without importing complete workflow state or provider + models. +4. Its reducer can update only its documented checkpoint fields and rejects stale or + malformed outcomes. +5. The station runs through the local fixture runner without LangGraph, Redis, Jira, or + GitHub. +6. Existing checkpoints resume and the feature, bug, and task-takeover golden paths remain + behaviorally equivalent. +7. The architecture report records a smaller dependency and state-access surface for the + migrated operation. diff --git a/src/forge/domain/__init__.py b/src/forge/domain/__init__.py new file mode 100644 index 00000000..0f090e67 --- /dev/null +++ b/src/forge/domain/__init__.py @@ -0,0 +1,39 @@ +"""Forge-owned contracts independent of workflow and provider runtimes.""" + +from forge.domain.commands import WorkflowCommand, WorkflowCommandType +from forge.domain.effects import EffectCommand, EffectResult, EffectResultStatus +from forge.domain.identity import ( + ResourceIdentity, + StationInvocationIdentity, + WorkflowIdentity, + stable_identity, +) +from forge.domain.observations import Observation, ObservationSource +from forge.domain.schema import DomainModel, JsonValue, VersionedDomainModel +from forge.domain.stations import ( + StationFailure, + StationOutcome, + StationOutcomeStatus, + StationRequest, +) + +__all__ = [ + "DomainModel", + "EffectCommand", + "EffectResult", + "EffectResultStatus", + "JsonValue", + "Observation", + "ObservationSource", + "ResourceIdentity", + "StationFailure", + "StationInvocationIdentity", + "StationOutcome", + "StationOutcomeStatus", + "StationRequest", + "VersionedDomainModel", + "WorkflowCommand", + "WorkflowCommandType", + "WorkflowIdentity", + "stable_identity", +] diff --git a/src/forge/domain/commands.py b/src/forge/domain/commands.py new file mode 100644 index 00000000..afbe3be1 --- /dev/null +++ b/src/forge/domain/commands.py @@ -0,0 +1,31 @@ +"""Commands requesting evaluation of a workflow instance.""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum + +from pydantic import Field + +from forge.domain.identity import WorkflowIdentity +from forge.domain.schema import JsonValue, VersionedDomainModel + + +class WorkflowCommandType(StrEnum): + START = "start" + RESUME = "resume" + APPROVE = "approve" + REJECT = "reject" + RETRY = "retry" + CANCEL = "cancel" + SYNCHRONIZE = "synchronize" + + +class WorkflowCommand(VersionedDomainModel): + command_id: str = Field(min_length=1) + command_type: WorkflowCommandType + workflow: WorkflowIdentity + requested_at: datetime + observation_ids: tuple[str, ...] = () + arguments: dict[str, JsonValue] = Field(default_factory=dict) + correlation: dict[str, JsonValue] = Field(default_factory=dict) diff --git a/src/forge/domain/effects.py b/src/forge/domain/effects.py new file mode 100644 index 00000000..f72cbb73 --- /dev/null +++ b/src/forge/domain/effects.py @@ -0,0 +1,39 @@ +"""Provider-neutral external-effect intent and result contracts.""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum + +from pydantic import Field + +from forge.domain.identity import ResourceIdentity, WorkflowIdentity +from forge.domain.schema import JsonValue, VersionedDomainModel + + +class EffectResultStatus(StrEnum): + SUCCEEDED = "succeeded" + PRECONDITION_FAILED = "precondition_failed" + RETRYABLE_FAILURE = "retryable_failure" + TERMINAL_FAILURE = "terminal_failure" + + +class EffectCommand(VersionedDomainModel): + effect_id: str = Field(min_length=1) + idempotency_key: str = Field(min_length=1) + workflow: WorkflowIdentity + operation: str = Field(min_length=1) + target: ResourceIdentity + expected_precondition: dict[str, JsonValue] = Field(default_factory=dict) + payload: dict[str, JsonValue] = Field(default_factory=dict) + + +class EffectResult(VersionedDomainModel): + effect_id: str = Field(min_length=1) + idempotency_key: str = Field(min_length=1) + status: EffectResultStatus + completed_at: datetime + provider_reference: str | None = None + output: dict[str, JsonValue] = Field(default_factory=dict) + error_code: str | None = None + error_message: str | None = None diff --git a/src/forge/domain/identity.py b/src/forge/domain/identity.py new file mode 100644 index 00000000..aab78646 --- /dev/null +++ b/src/forge/domain/identity.py @@ -0,0 +1,35 @@ +"""Stable identities shared by workflow, station and effect contracts.""" + +from __future__ import annotations + +import hashlib +import json + +from pydantic import Field + +from forge.domain.schema import DomainModel, JsonValue + + +def stable_identity(namespace: str, parts: dict[str, JsonValue]) -> str: + """Derive a deterministic identity from canonical JSON data.""" + encoded = json.dumps(parts, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + digest = hashlib.sha256(encoded.encode()).hexdigest() + return f"{namespace}:{digest}" + + +class WorkflowIdentity(DomainModel): + run_id: str = Field(min_length=1) + workflow_name: str = Field(min_length=1) + definition_revision: int = Field(ge=1) + definition_digest: str | None = None + + +class ResourceIdentity(DomainModel): + resource_type: str = Field(min_length=1) + external_id: str = Field(min_length=1) + namespace: str | None = None + + +class StationInvocationIdentity(DomainModel): + invocation_id: str = Field(min_length=1) + station_name: str = Field(min_length=1) diff --git a/src/forge/domain/observations.py b/src/forge/domain/observations.py new file mode 100644 index 00000000..f668d3a9 --- /dev/null +++ b/src/forge/domain/observations.py @@ -0,0 +1,30 @@ +"""Observations of external state supplied through any ingress path.""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum + +from pydantic import Field + +from forge.domain.identity import ResourceIdentity +from forge.domain.schema import JsonValue, VersionedDomainModel + + +class ObservationSource(StrEnum): + WEBHOOK = "webhook" + POLLER = "poller" + INTERNAL = "internal" + + +class Observation(VersionedDomainModel): + observation_id: str = Field(min_length=1) + source: ObservationSource + source_system: str = Field(min_length=1) + resource: ResourceIdentity + resource_revision: str | None = None + observed_at: datetime + received_at: datetime + facts: dict[str, JsonValue] = Field(default_factory=dict) + correlation: dict[str, JsonValue] = Field(default_factory=dict) + evidence_reference: str | None = None diff --git a/src/forge/domain/schema.py b/src/forge/domain/schema.py new file mode 100644 index 00000000..2c542a29 --- /dev/null +++ b/src/forge/domain/schema.py @@ -0,0 +1,23 @@ +"""Shared validation and serialization rules for Forge domain contracts.""" + +from __future__ import annotations + +from typing import Literal + +from pydantic import BaseModel, ConfigDict +from typing_extensions import TypeAliasType + +JsonScalar = str | int | float | bool | None +JsonValue = TypeAliasType("JsonValue", JsonScalar | list["JsonValue"] | dict[str, "JsonValue"]) + + +class DomainModel(BaseModel): + """Strict, immutable and JSON-safe base for durable domain messages.""" + + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + +class VersionedDomainModel(DomainModel): + """Base for the first version of Forge-owned runtime contracts.""" + + schema_version: Literal["1.0"] = "1.0" diff --git a/src/forge/domain/stations.py b/src/forge/domain/stations.py new file mode 100644 index 00000000..583d6b3a --- /dev/null +++ b/src/forge/domain/stations.py @@ -0,0 +1,56 @@ +"""Typed invocation and outcome contracts for independently runnable stations.""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum +from typing import Generic, TypeVar + +from pydantic import Field + +from forge.domain.effects import EffectCommand +from forge.domain.identity import StationInvocationIdentity, WorkflowIdentity +from forge.domain.schema import DomainModel, JsonValue, VersionedDomainModel + +InputT = TypeVar("InputT", bound=DomainModel) +OutputT = TypeVar("OutputT", bound=DomainModel) + + +class StationOutcomeStatus(StrEnum): + SUCCEEDED = "succeeded" + BLOCKED = "blocked" + WAITING = "waiting" + RETRYABLE_FAILURE = "retryable_failure" + TERMINAL_FAILURE = "terminal_failure" + + +class StationFailure(DomainModel): + code: str = Field(min_length=1) + message: str = Field(min_length=1) + details: dict[str, JsonValue] = Field(default_factory=dict) + + +class StationRequest(VersionedDomainModel, Generic[InputT]): + workflow: WorkflowIdentity + invocation: StationInvocationIdentity + contract_name: str = Field(min_length=1) + contract_version: str = Field(min_length=1) + attempt: int = Field(ge=1) + requested_at: datetime + deadline: datetime | None = None + artifact_references: tuple[str, ...] = () + policy_context: dict[str, JsonValue] = Field(default_factory=dict) + input: InputT + + +class StationOutcome(VersionedDomainModel, Generic[OutputT]): + workflow: WorkflowIdentity + invocation: StationInvocationIdentity + contract_name: str = Field(min_length=1) + contract_version: str = Field(min_length=1) + status: StationOutcomeStatus + completed_at: datetime + output: OutputT | None = None + requested_effects: tuple[EffectCommand, ...] = () + reason: str | None = None + failure: StationFailure | None = None diff --git a/tests/unit/domain/test_architecture.py b/tests/unit/domain/test_architecture.py new file mode 100644 index 00000000..79333d69 --- /dev/null +++ b/tests/unit/domain/test_architecture.py @@ -0,0 +1,31 @@ +"""Dependency-direction checks for the provider-independent domain package.""" + +import ast +from pathlib import Path + +DOMAIN_ROOT = Path(__file__).parents[3] / "src" / "forge" / "domain" +PROHIBITED_PREFIXES = ( + "langgraph", + "redis", + "forge.integrations", + "forge.orchestrator", + "forge.queue", + "forge.workflow", +) + + +def test_domain_contracts_do_not_import_runtime_or_provider_packages() -> None: + violations: list[str] = [] + for path in sorted(DOMAIN_ROOT.glob("*.py")): + tree = ast.parse(path.read_text(), filename=str(path)) + for node in ast.walk(tree): + modules: list[str] = [] + if isinstance(node, ast.Import): + modules = [alias.name for alias in node.names] + elif isinstance(node, ast.ImportFrom) and node.module: + modules = [node.module] + for module in modules: + if module.startswith(PROHIBITED_PREFIXES): + violations.append(f"{path.name}:{node.lineno}: {module}") + + assert not violations, "Prohibited domain dependencies:\n" + "\n".join(violations) diff --git a/tests/unit/domain/test_contracts.py b/tests/unit/domain/test_contracts.py new file mode 100644 index 00000000..9bc0d0e8 --- /dev/null +++ b/tests/unit/domain/test_contracts.py @@ -0,0 +1,119 @@ +"""Conformance tests for the Phase 1 domain-contract kernel.""" + +from datetime import UTC, datetime + +import pytest +from pydantic import ValidationError + +from forge.domain import ( + DomainModel, + Observation, + ObservationSource, + ResourceIdentity, + StationInvocationIdentity, + StationOutcome, + StationOutcomeStatus, + StationRequest, + WorkflowIdentity, + stable_identity, +) + +NOW = datetime(2026, 8, 27, tzinfo=UTC) + + +class ExampleInput(DomainModel): + ticket_key: str + + +class ExampleOutput(DomainModel): + summary: str + + +def workflow_identity() -> WorkflowIdentity: + return WorkflowIdentity( + run_id="run-1", + workflow_name="feature", + definition_revision=3, + ) + + +def test_observation_round_trips_through_json() -> None: + observation = Observation( + observation_id="github:event-1", + source=ObservationSource.WEBHOOK, + source_system="github", + resource=ResourceIdentity(resource_type="pull_request", external_id="acme/repo#7"), + resource_revision="abc123", + observed_at=NOW, + received_at=NOW, + facts={"merged": False, "labels": ["ready"]}, + ) + + restored = Observation.model_validate_json(observation.model_dump_json()) + + assert restored == observation + + +def test_contracts_reject_unknown_fields_and_statuses() -> None: + with pytest.raises(ValidationError): + ResourceIdentity( + resource_type="issue", + external_id="TEST-1", + provider="jira", # type: ignore[call-arg] + ) + + with pytest.raises(ValidationError): + StationOutcome[ExampleOutput]( + workflow=workflow_identity(), + invocation=StationInvocationIdentity( + invocation_id="invocation-1", station_name="example" + ), + contract_name="example", + contract_version="1.0", + status="maybe", # type: ignore[arg-type] + completed_at=NOW, + ) + + with pytest.raises(ValidationError): + Observation( + schema_version="2.0", # type: ignore[arg-type] + observation_id="future", + source=ObservationSource.INTERNAL, + source_system="forge", + resource=ResourceIdentity(resource_type="issue", external_id="TEST-1"), + observed_at=NOW, + received_at=NOW, + ) + + +def test_station_request_and_outcome_are_typed_and_round_trip() -> None: + invocation = StationInvocationIdentity(invocation_id="invocation-1", station_name="example") + request = StationRequest[ExampleInput]( + workflow=workflow_identity(), + invocation=invocation, + contract_name="example", + contract_version="1.0", + attempt=1, + requested_at=NOW, + input=ExampleInput(ticket_key="TEST-1"), + ) + outcome = StationOutcome[ExampleOutput]( + workflow=request.workflow, + invocation=request.invocation, + contract_name=request.contract_name, + contract_version=request.contract_version, + status=StationOutcomeStatus.SUCCEEDED, + completed_at=NOW, + output=ExampleOutput(summary="done"), + ) + + assert StationRequest[ExampleInput].model_validate_json(request.model_dump_json()) == request + assert StationOutcome[ExampleOutput].model_validate_json(outcome.model_dump_json()) == outcome + + +def test_stable_identity_is_order_independent_and_namespaced() -> None: + first = stable_identity("observation", {"provider": "github", "event": 7}) + second = stable_identity("observation", {"event": 7, "provider": "github"}) + + assert first == second + assert first.startswith("observation:") From 6c7fe3635fd54881382fcf9ddfd2d9ee41e02ad7 Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Thu, 27 Aug 2026 15:56:34 +0300 Subject: [PATCH 2/4] feat: adapt source control events to observations --- .../source_control/observations.py | 98 +++++++++++++++++++ .../source_control/test_observations.py | 65 ++++++++++++ 2 files changed, 163 insertions(+) create mode 100644 src/forge/integrations/source_control/observations.py create mode 100644 tests/unit/integrations/source_control/test_observations.py diff --git a/src/forge/integrations/source_control/observations.py b/src/forge/integrations/source_control/observations.py new file mode 100644 index 00000000..32f8b269 --- /dev/null +++ b/src/forge/integrations/source_control/observations.py @@ -0,0 +1,98 @@ +"""Compatibility adapter from source-control events to Forge observations.""" + +from __future__ import annotations + +from dataclasses import asdict, is_dataclass +from datetime import datetime +from enum import Enum +from typing import Any + +from forge.domain import ( + JsonValue, + Observation, + ObservationSource, + ResourceIdentity, + stable_identity, +) +from forge.integrations.source_control.contracts import NormalizedEvent + + +def _json_value(value: Any) -> JsonValue: + if is_dataclass(value) and not isinstance(value, type): + return _json_value(asdict(value)) + if isinstance(value, Enum): + return str(value.value) + if isinstance(value, datetime): + return value.isoformat() + if isinstance(value, dict): + return {str(key): _json_value(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_json_value(item) for item in value] + if value is None or isinstance(value, (str, int, float, bool)): + return value + raise TypeError(f"Unsupported observation fact type: {type(value).__name__}") + + +def normalized_event_to_observation( + event: NormalizedEvent, + *, + source: ObservationSource = ObservationSource.WEBHOOK, +) -> Observation: + """Convert an existing transport event without retaining its raw payload.""" + change_request = event.change_request + native_id = change_request.identity.native_id if change_request else None + external_id = event.repo_ref.id + resource_type = "repository" + revision: str | None = None + if change_request: + resource_type = "change_request" + external_id = f"{event.repo_ref.id}#{native_id}" + revision = change_request.head_sha or None + elif event.check: + resource_type = "check" + external_id = f"{event.repo_ref.id}:{event.check.name}" + revision = f"{event.check.status.value}:{event.check.conclusion.value}" + elif event.comment: + resource_type = "comment" + external_id = f"{event.repo_ref.id}:{event.comment.id}" + elif event.review: + resource_type = "review" + external_id = f"{event.repo_ref.id}:{event.review.id}" + + facts = _json_value( + { + "kind": event.kind, + "repository": event.repo_ref, + "actor": event.actor, + "change_request": event.change_request, + "comment": event.comment, + "review": event.review, + "check": event.check, + "check_suite_status": event.check_suite_status, + } + ) + assert isinstance(facts, dict) + observation_id = stable_identity( + "observation", + { + "source_system": event.repo_ref.provider.value, + "event_id": event.id, + "resource_revision": revision, + }, + ) + return Observation( + observation_id=observation_id, + source=source, + source_system=event.repo_ref.provider.value, + resource=ResourceIdentity( + resource_type=resource_type, + external_id=external_id, + namespace=event.repo_ref.connection, + ), + resource_revision=revision, + observed_at=event.received_at, + received_at=event.received_at, + facts=facts, + correlation={"transport_event_id": event.id, "repository_id": event.repo_ref.id}, + evidence_reference=f"source-control-event:{event.id}" if event.raw else None, + ) diff --git a/tests/unit/integrations/source_control/test_observations.py b/tests/unit/integrations/source_control/test_observations.py new file mode 100644 index 00000000..9bc7f1bf --- /dev/null +++ b/tests/unit/integrations/source_control/test_observations.py @@ -0,0 +1,65 @@ +from datetime import UTC, datetime + +from forge.domain import Observation, ObservationSource +from forge.integrations.source_control.contracts import ( + Actor, + ChangeRequest, + ChangeRequestIdentity, + ChangeRequestState, + EventKind, + NormalizedEvent, + Provider, + RepositoryRef, +) +from forge.integrations.source_control.observations import normalized_event_to_observation + + +def _event() -> NormalizedEvent: + repo = RepositoryRef( + id="acme/api", + provider=Provider.GITHUB, + connection="public", + namespace="acme", + default_branch="main", + change_request_mode="direct", + ) + return NormalizedEvent( + id="delivery-7", + kind=EventKind.CR_UPDATED, + repo_ref=repo, + actor=Actor(login="octocat", is_bot=False), + received_at=datetime(2026, 8, 27, tzinfo=UTC), + change_request=ChangeRequest( + identity=ChangeRequestIdentity( + connection="public", repository_id="acme/api", native_id=42 + ), + url="https://github.com/acme/api/pull/42", + title="Change", + body="Body", + state=ChangeRequestState.OPEN, + source_branch="feature", + target_branch="main", + head_sha="abc123", + ), + raw={"provider": "payload is retained outside the domain contract"}, + ) + + +def test_conversion_is_deterministic_and_json_round_trippable() -> None: + first = normalized_event_to_observation(_event()) + second = normalized_event_to_observation(_event()) + + assert first == second + assert first.resource.external_id == "acme/api#42" + assert first.resource_revision == "abc123" + assert first.facts["kind"] == "cr_updated" + assert "raw" not in first.facts + assert Observation.model_validate_json(first.model_dump_json()) == first + + +def test_poller_and_webhook_use_same_identity_for_same_external_event() -> None: + webhook = normalized_event_to_observation(_event()) + polled = normalized_event_to_observation(_event(), source=ObservationSource.POLLER) + + assert webhook.observation_id == polled.observation_id + assert polled.source is ObservationSource.POLLER From f947084dc5a121705a38b7f97253a891b8da015c Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Thu, 27 Aug 2026 15:56:34 +0300 Subject: [PATCH 3/4] feat: run implementation resolution as typed station --- .../phase-1-domain-contracts-plan.md | 27 +- src/forge/workflow/implementation_input.py | 233 +++--------------- src/forge/workflow/projections/__init__.py | 1 + .../projections/implementation_input.py | 145 +++++++++++ src/forge/workflow/reducers/__init__.py | 1 + .../workflow/reducers/implementation_input.py | 52 ++++ src/forge/workflow/stations/__init__.py | 1 + .../workflow/stations/implementation_input.py | 165 +++++++++++++ src/forge/workflow/stations/runner.py | 34 +++ tests/unit/domain/test_architecture.py | 15 ++ .../test_implementation_input_station.py | 93 +++++++ 11 files changed, 559 insertions(+), 208 deletions(-) create mode 100644 src/forge/workflow/projections/__init__.py create mode 100644 src/forge/workflow/projections/implementation_input.py create mode 100644 src/forge/workflow/reducers/__init__.py create mode 100644 src/forge/workflow/reducers/implementation_input.py create mode 100644 src/forge/workflow/stations/__init__.py create mode 100644 src/forge/workflow/stations/implementation_input.py create mode 100644 src/forge/workflow/stations/runner.py create mode 100644 tests/unit/workflow/test_implementation_input_station.py diff --git a/docs/architecture/phase-1-domain-contracts-plan.md b/docs/architecture/phase-1-domain-contracts-plan.md index 6943b09f..b3c170b4 100644 --- a/docs/architecture/phase-1-domain-contracts-plan.md +++ b/docs/architecture/phase-1-domain-contracts-plan.md @@ -1,6 +1,6 @@ # Phase 1 implementation plan: versioned domain contracts -**Status:** Proposed +**Status:** Implemented in PR 324 **Depends on:** Phase 0 baseline and the stacked integration of `dev`, PR 317, and PR 318 @@ -167,19 +167,28 @@ outcomes in tests; do not dual-execute external effects. **Why it matters:** the phase ends with an independently testable station and measurable coupling reduction, not only unused model classes. -## PR sequence +## Delivery -Keep Phase 1 reviewable as four stacked PRs on top of PR 318 (or its eventual merged -successor): +Phase 1 was delivered as one isolated PR stacked on PR 318, with the internal stages kept +as reviewable commits and package boundaries: 1. **Contract kernel and architecture rule** — Stage 1.1. 2. **Observation adapter** — Stage 1.2. -3. **Projection/reducer framework** — Stage 1.3. -4. **Implementation-input station and conformance runner** — Stages 1.4 and 1.5. +3. **Projection/reducer framework and implementation-input station** — Stages 1.3–1.4. +4. **Conformance runner, characterization, and measurements** — Stage 1.5. -Each PR must pass the full current gate. Contract additions must not require checkpoint -migration, and the final PR must run feature, bug, and task-takeover characterization -tests. +The PR requires no checkpoint migration and runs feature, bug, and task-takeover +characterization tests. + +### Resulting coupling measures + +- The compatibility facade decreased from 253 lines to 86 lines. +- The station receives nine explicitly scoped input groups and writes no checkpoint state. +- The reducer owns exactly four checkpoint fields: `artifacts`, `work_units`, + `current_work_unit_id`, and `work_resolution`. +- The station imports no Jira/GitHub provider, LangGraph type, `BaseState`, worker, or queue. +- The same serialized request runs through the local runner without Redis, LangGraph, Jira, + or source-control clients. ## Implications diff --git a/src/forge/workflow/implementation_input.py b/src/forge/workflow/implementation_input.py index f5af8aa5..4840823d 100644 --- a/src/forge/workflow/implementation_input.py +++ b/src/forge/workflow/implementation_input.py @@ -1,20 +1,18 @@ -"""Resolve workflow-specific planning state into repository-scoped implementation input.""" +"""Compatibility facade for contract-backed implementation-input resolution.""" from __future__ import annotations from collections.abc import Mapping from dataclasses import dataclass -from typing import Any, Literal, Protocol +from typing import Any, Protocol, cast -from forge.integrations.jira.models import JiraIssue from forge.workflow.base import ArtifactRef, WorkUnit -from forge.workflow.planning_state import ( - artifact_is_current, - content_digest, - planning_artifacts, +from forge.workflow.projections.implementation_input import project_implementation_input +from forge.workflow.reducers.implementation_input import reduce_implementation_input +from forge.workflow.stations.implementation_input import ( + NoPendingImplementationWork as StationNoPendingImplementationWork, ) - -ArtifactKind = Literal["task", "epic_plan", "plan", "spec", "rca", "prd", "ticket"] +from forge.workflow.stations.implementation_input import run_implementation_input_station class NoPendingImplementationWork(Exception): @@ -22,38 +20,38 @@ class NoPendingImplementationWork(Exception): class IssueReader(Protocol): - """The small Jira API surface needed by the resolver.""" + """Small external-fact surface used only by the request projector.""" - async def get_issue(self, issue_key: str) -> JiraIssue: ... + async def get_issue(self, issue_key: str) -> Any: ... @dataclass(frozen=True) class ResolvedImplementationInput: - """Normalized input shared by Feature, Bug, and Task-takeover execution.""" + """Backward-compatible view of the typed station outcome.""" work_unit: WorkUnit context_artifacts: tuple[ArtifactRef, ...] instructions: str summary: str | None = None + _station_request: Any | None = None + _station_outcome: Any | None = None def state_update(self, state: Mapping[str, Any] | None = None) -> dict[str, Any]: - """Return checkpoint-safe normalized state fields for this decision. + if self._station_request is not None and self._station_outcome is not None: + return reduce_implementation_input( + state or {}, self._station_request, self._station_outcome + ) - Existing artifacts and work units are retained by identity so resolution - across repositories or retries builds an audit trail instead of replacing it. - """ existing_artifacts = list((state or {}).get("artifacts") or []) artifacts_by_id = {artifact.get("id"): artifact for artifact in existing_artifacts} for artifact in self.context_artifacts: artifacts_by_id[artifact.get("id")] = artifact - existing_units = list((state or {}).get("work_units") or []) units_by_id = {unit.get("id"): unit for unit in existing_units} previous = units_by_id.get(self.work_unit["id"]) - if previous and previous.get("status") == "completed": - units_by_id[self.work_unit["id"]] = previous - else: - units_by_id[self.work_unit["id"]] = self.work_unit + units_by_id[self.work_unit["id"]] = ( + previous if previous and previous.get("status") == "completed" else self.work_unit + ) return { "artifacts": list(artifacts_by_id.values()), "work_units": list(units_by_id.values()), @@ -66,186 +64,23 @@ def state_update(self, state: Mapping[str, Any] | None = None) -> dict[str, Any] } -def _digest(content: str) -> str: - return content_digest(content) - - -def _repo_labels(issue: JiraIssue) -> set[str]: - return {label.removeprefix("repo:") for label in issue.labels if label.startswith("repo:")} - - -def _content(issue: JiraIssue) -> str: - return issue.description.strip() - - -def _issue_artifact(kind: ArtifactKind, issue: JiraIssue, repository: str) -> ArtifactRef: - content = _content(issue) - return { - "id": f"jira:{issue.key}:{kind}", - "kind": kind, - "source": issue.key, - "content": content, - "digest": _digest(content), - "repository": repository, - } - - -def _assert_issue_repository(issue: JiraIssue, repository: str) -> None: - repos = _repo_labels(issue) - if repos and repository not in repos: - raise ValueError( - f"Jira issue {issue.key} is scoped to {sorted(repos)}, not current repository {repository}" - ) - - -def _task_candidates(state: Mapping[str, Any], repository: str) -> list[str]: - implemented = set(state.get("implemented_tasks") or []) - implemented.update( - unit.get("id", "") - for unit in state.get("work_units") or [] - if unit.get("status") == "completed" - ) - mapped = state.get("tasks_by_repo") or {} - candidates: list[str] = [] - current = state.get("current_task_key") - if isinstance(current, str) and current and current not in implemented: - for other_repo, keys in mapped.items(): - if other_repo != repository and current in (keys or []): - raise ValueError(f"Current task {current} belongs to repository {other_repo}") - candidates.append(current) - for key in mapped.get(repository, []): - if isinstance(key, str) and key not in implemented and key not in candidates: - candidates.append(key) - for unit in state.get("work_units") or []: - if ( - unit.get("kind") == "task" - and unit.get("repository") == repository - and unit.get("status") in {"pending", "active"} - ): - key = unit.get("jira_key") or unit.get("key") or unit.get("id") - if isinstance(key, str) and key not in implemented and key not in candidates: - candidates.append(key) - ticket_type = state.get("ticket_type") - ticket_type_name = getattr(ticket_type, "value", ticket_type) - ticket_key = state.get("ticket_key") - if ( - ticket_type_name in {"Task", "Epic"} - and isinstance(ticket_key, str) - and ticket_key not in implemented - and ticket_key not in candidates - ): - candidates.append(ticket_key) - return candidates - - async def resolve_implementation_input( state: Mapping[str, Any], jira: IssueReader ) -> ResolvedImplementationInput: - """Resolve task-first implementation input, strictly scoped to ``current_repo``. - - Primary work precedence is current Task, the first pending repository Task, - repository Epic plan, plan, spec, RCA, PRD, then the root ticket. Lower-level - available artifacts are retained as ordered context rather than discarded. - """ - repository = state.get("current_repository") or state.get("current_repo") - if not isinstance(repository, str) or not repository.strip(): - raise ValueError("current_repo is required to resolve implementation input") - repository = repository.strip() - - artifacts: list[ArtifactRef] = [] - summaries: dict[str, str] = {} - stale_tasks = [ - unit.get("id") - for unit in state.get("work_units") or [] - if unit.get("kind") == "task" - and unit.get("repository") == repository - and unit.get("status") == "stale" - ] - if stale_tasks: - raise ValueError( - f"Repository {repository} has Tasks derived from stale planning: {stale_tasks}" - ) - task_keys = _task_candidates(state, repository) - repository_tasks = (state.get("tasks_by_repo") or {}).get(repository, []) - if repository_tasks and not task_keys: - raise NoPendingImplementationWork(f"All Jira tasks are complete for {repository}") - for task_key in task_keys[:1]: - issue = await jira.get_issue(task_key) - _assert_issue_repository(issue, repository) - artifact = _issue_artifact("task", issue, repository) - if not artifact["content"] and issue.summary.strip(): - artifact["content"] = issue.summary.strip() - artifact["digest"] = _digest(artifact["content"]) - if artifact["content"]: - artifacts.append(artifact) - summaries[artifact["id"]] = issue.summary - - for epic_key in state.get("epic_keys") or []: - if not isinstance(epic_key, str): - continue - issue = await jira.get_issue(epic_key) - # Epic plans are never treated as global: they must explicitly name this repo. - if repository not in _repo_labels(issue): - continue - artifact = _issue_artifact("epic_plan", issue, repository) - if artifact["content"]: - artifacts.append(artifact) - summaries[artifact["id"]] = issue.summary - - existing_ids = {artifact.get("id") for artifact in artifacts} - rank = {"task": 0, "epic_plan": 1, "plan": 2, "spec": 3, "rca": 4, "prd": 5, "ticket": 6} - layered = sorted( - planning_artifacts(state), - key=lambda artifact: rank.get(str(artifact.get("kind")), 99), - ) - for artifact in layered: - if artifact.get("id") in existing_ids or not artifact_is_current(artifact): - continue - artifact_repo = artifact.get("repository") - if artifact_repo not in {None, repository}: - continue - if artifact.get("content"): - artifacts.append(artifact) - existing_ids.add(artifact.get("id")) - - ticket_key = state.get("ticket_key") - if isinstance(ticket_key, str) and ticket_key and ticket_key not in task_keys[:1]: - issue = await jira.get_issue(ticket_key) - _assert_issue_repository(issue, repository) - artifact = _issue_artifact("ticket", issue, repository) - if artifact["content"]: - artifacts.append(artifact) - summaries[artifact["id"]] = issue.summary - - if not artifacts: - raise ValueError(f"No implementation artifact is available for repository {repository}") - - primary = artifacts[0] - jira_key = primary.get("jira_key") - if not jira_key and primary.get("kind") in {"task", "epic_plan", "ticket"}: - source = primary.get("source") - jira_key = source if isinstance(source, str) else None - work_id = jira_key or f"internal:{repository}:{primary['kind']}:{primary['digest'][7:19]}" - completed_ids = set(state.get("implemented_tasks") or []) - completed_ids.update( - unit.get("id", "") - for unit in state.get("work_units") or [] - if unit.get("status") == "completed" - ) - if work_id in completed_ids: - raise NoPendingImplementationWork(f"Work unit {work_id} is already complete") - work_unit: WorkUnit = { - "id": work_id, - "kind": primary["kind"], - "key": jira_key, - "repository": repository, - "status": "pending", - "source_artifact_ids": [primary["id"]], - "context_artifact_ids": [artifact["id"] for artifact in artifacts[1:]], - } + """Project a checkpoint, invoke the station, and expose its compatible result.""" + try: + request = await project_implementation_input(state, jira) + outcome = run_implementation_input_station(request) + except StationNoPendingImplementationWork as exc: + raise NoPendingImplementationWork(str(exc)) from exc + assert outcome.output is not None return ResolvedImplementationInput( - work_unit=work_unit, - context_artifacts=tuple(artifacts), - instructions=primary["content"], - summary=summaries.get(primary["id"]), + work_unit=cast(WorkUnit, outcome.output.work_unit), + context_artifacts=tuple( + cast(ArtifactRef, item) for item in outcome.output.context_artifacts + ), + instructions=outcome.output.instructions, + summary=outcome.output.summary, + _station_request=request, + _station_outcome=outcome, ) diff --git a/src/forge/workflow/projections/__init__.py b/src/forge/workflow/projections/__init__.py new file mode 100644 index 00000000..2e5c9213 --- /dev/null +++ b/src/forge/workflow/projections/__init__.py @@ -0,0 +1 @@ +"""Adapters from checkpoint/provider facts to station requests.""" diff --git a/src/forge/workflow/projections/implementation_input.py b/src/forge/workflow/projections/implementation_input.py new file mode 100644 index 00000000..45c77dfe --- /dev/null +++ b/src/forge/workflow/projections/implementation_input.py @@ -0,0 +1,145 @@ +"""Projection for the implementation-input station.""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import UTC, datetime +from typing import Any, Protocol + +from forge.domain import ( + StationInvocationIdentity, + StationRequest, + WorkflowIdentity, + stable_identity, +) +from forge.workflow.planning_state import planning_artifacts +from forge.workflow.stations.implementation_input import ( + CONTRACT_NAME, + CONTRACT_VERSION, + ImplementationInput, + NoPendingImplementationWork, + WorkItemSnapshot, +) + + +class IssueSnapshotReader(Protocol): + async def get_issue(self, issue_key: str) -> Any: ... + + +def _task_candidates(state: Mapping[str, Any], repository: str) -> list[str]: + completed = set(state.get("implemented_tasks") or []) | { + str(unit.get("id")) + for unit in state.get("work_units") or [] + if unit.get("status") == "completed" + } + mapped = state.get("tasks_by_repo") or {} + candidates: list[str] = [] + current = state.get("current_task_key") + if isinstance(current, str) and current and current not in completed: + for other_repo, keys in mapped.items(): + if other_repo != repository and current in (keys or []): + raise ValueError(f"Current task {current} belongs to repository {other_repo}") + candidates.append(current) + for key in mapped.get(repository, []): + if isinstance(key, str) and key not in completed and key not in candidates: + candidates.append(key) + for unit in state.get("work_units") or []: + if ( + unit.get("kind") == "task" + and unit.get("repository") == repository + and unit.get("status") in {"pending", "active"} + ): + key = unit.get("jira_key") or unit.get("key") or unit.get("id") + if isinstance(key, str) and key not in completed and key not in candidates: + candidates.append(key) + ticket_type = getattr(state.get("ticket_type"), "value", state.get("ticket_type")) + ticket_key = state.get("ticket_key") + if ( + ticket_type in {"Task", "Epic"} + and isinstance(ticket_key, str) + and ticket_key not in completed + and ticket_key not in candidates + ): + candidates.append(ticket_key) + return candidates + + +async def project_implementation_input( + state: Mapping[str, Any], reader: IssueSnapshotReader +) -> StationRequest[ImplementationInput]: + repository = state.get("current_repository") or state.get("current_repo") + if not isinstance(repository, str) or not repository.strip(): + raise ValueError("current_repo is required to resolve implementation input") + repository = repository.strip() + candidates = _task_candidates(state, repository) + stale = [ + unit.get("id") + for unit in state.get("work_units") or [] + if unit.get("kind") == "task" + and unit.get("repository") == repository + and unit.get("status") == "stale" + ] + if stale: + raise ValueError(f"Repository {repository} has Tasks derived from stale planning: {stale}") + if (state.get("tasks_by_repo") or {}).get(repository) and not candidates: + raise NoPendingImplementationWork(f"All Jira tasks are complete for {repository}") + ticket_key = state.get("ticket_key") if isinstance(state.get("ticket_key"), str) else None + epic_keys = tuple(key for key in state.get("epic_keys") or [] if isinstance(key, str)) + keys = list(dict.fromkeys([*candidates[:1], *epic_keys, *([ticket_key] if ticket_key else [])])) + items: dict[str, WorkItemSnapshot] = {} + for key in keys: + issue = await reader.get_issue(key) + items[key] = WorkItemSnapshot( + # The requested key is authoritative; lightweight reader doubles and + # some provider responses do not repeat it on the returned object. + key=key, + summary=issue.summary or "", + description=issue.description or "", + labels=tuple(issue.labels or []), + ) + completed = set(state.get("implemented_tasks") or []) | { + str(unit.get("id")) + for unit in state.get("work_units") or [] + if unit.get("status") == "completed" + } + run_id = str(state.get("thread_id") or ticket_key or "local") + workflow_name = str(state.get("workflow_name") or state.get("ticket_type") or "legacy") + revision = int(state.get("workflow_revision") or 1) + invocation_id = stable_identity( + "station-invocation", + {"run_id": run_id, "station": CONTRACT_NAME, "repository": repository}, + ) + timestamp = ( + datetime.fromisoformat(str(state["updated_at"])) + if state.get("updated_at") + else datetime(1970, 1, 1, tzinfo=UTC) + ) + return StationRequest[ImplementationInput]( + workflow=WorkflowIdentity( + run_id=run_id, + workflow_name=workflow_name, + definition_revision=revision, + definition_digest=state.get("workflow_digest"), + ), + invocation=StationInvocationIdentity( + invocation_id=invocation_id, station_name=CONTRACT_NAME + ), + contract_name=CONTRACT_NAME, + contract_version=CONTRACT_VERSION, + attempt=int(state.get("retry_count") or 0) + 1, + requested_at=timestamp, + artifact_references=tuple( + str(item.get("id")) for item in planning_artifacts(state) if item.get("id") + ), + input=ImplementationInput( + repository=repository, + ticket_key=ticket_key, + candidate_task_keys=tuple(candidates), + configured_repository_tasks=bool((state.get("tasks_by_repo") or {}).get(repository)), + epic_keys=epic_keys, + artifacts=tuple(planning_artifacts(state)), + work_units=tuple(state.get("work_units") or []), + implemented_work_ids=tuple(sorted(completed)), + work_items=items, + ), + ) diff --git a/src/forge/workflow/reducers/__init__.py b/src/forge/workflow/reducers/__init__.py new file mode 100644 index 00000000..99426513 --- /dev/null +++ b/src/forge/workflow/reducers/__init__.py @@ -0,0 +1 @@ +"""Validated station-outcome reducers.""" diff --git a/src/forge/workflow/reducers/implementation_input.py b/src/forge/workflow/reducers/implementation_input.py new file mode 100644 index 00000000..dc49e514 --- /dev/null +++ b/src/forge/workflow/reducers/implementation_input.py @@ -0,0 +1,52 @@ +"""Allowlisted checkpoint reducer for implementation-input outcomes.""" + +from collections.abc import Mapping +from typing import Any + +from forge.domain import StationOutcome, StationOutcomeStatus, StationRequest +from forge.workflow.stations.implementation_input import ImplementationInput, ImplementationOutput + + +def reduce_implementation_input( + state: Mapping[str, Any], + request: StationRequest[ImplementationInput], + outcome: StationOutcome[ImplementationOutput], +) -> dict[str, Any]: + expected_run = state.get("thread_id") + if expected_run and expected_run != request.workflow.run_id: + raise ValueError("Station request does not belong to the checkpoint workflow run") + expected_name = state.get("workflow_name") + if expected_name and expected_name != request.workflow.workflow_name: + raise ValueError("Station request workflow definition does not match the checkpoint") + expected_revision = state.get("workflow_revision") + if expected_revision and expected_revision != request.workflow.definition_revision: + raise ValueError("Station request workflow revision does not match the checkpoint") + if outcome.workflow != request.workflow or outcome.invocation != request.invocation: + raise ValueError("Station outcome does not belong to this workflow invocation") + if (outcome.contract_name, outcome.contract_version) != ( + request.contract_name, + request.contract_version, + ): + raise ValueError("Station outcome contract does not match its request") + if outcome.status is not StationOutcomeStatus.SUCCEEDED or outcome.output is None: + raise ValueError(f"Implementation-input station did not succeed: {outcome.status}") + output = outcome.output + artifacts_by_id = {item.get("id"): dict(item) for item in state.get("artifacts") or []} + for artifact in output.context_artifacts: + artifacts_by_id[artifact.get("id")] = dict(artifact) + units_by_id = {item.get("id"): dict(item) for item in state.get("work_units") or []} + work_unit = dict(output.work_unit) + previous = units_by_id.get(work_unit["id"]) + units_by_id[work_unit["id"]] = ( + previous if previous and previous.get("status") == "completed" else work_unit + ) + return { + "artifacts": list(artifacts_by_id.values()), + "work_units": list(units_by_id.values()), + "current_work_unit_id": work_unit["id"], + "work_resolution": { + "strategy": "task_first", + "selected_work_unit_id": work_unit["id"], + "selected_artifact_id": work_unit["source_artifact_ids"][0], + }, + } diff --git a/src/forge/workflow/stations/__init__.py b/src/forge/workflow/stations/__init__.py new file mode 100644 index 00000000..adbaa459 --- /dev/null +++ b/src/forge/workflow/stations/__init__.py @@ -0,0 +1 @@ +"""Provider-independent station implementations.""" diff --git a/src/forge/workflow/stations/implementation_input.py b/src/forge/workflow/stations/implementation_input.py new file mode 100644 index 00000000..03f790df --- /dev/null +++ b/src/forge/workflow/stations/implementation_input.py @@ -0,0 +1,165 @@ +"""Pure station for selecting repository-scoped implementation work.""" + +from __future__ import annotations + +import hashlib + +from pydantic import Field + +from forge.domain import ( + DomainModel, + JsonValue, + StationOutcome, + StationOutcomeStatus, + StationRequest, +) + +CONTRACT_NAME = "implementation-input" +CONTRACT_VERSION = "1.0" + + +class NoPendingImplementationWork(Exception): + """Known implementation work has already completed.""" + + +class WorkItemSnapshot(DomainModel): + key: str + summary: str = "" + description: str = "" + labels: tuple[str, ...] = () + + +class ImplementationInput(DomainModel): + repository: str + ticket_key: str | None = None + candidate_task_keys: tuple[str, ...] = () + configured_repository_tasks: bool = False + epic_keys: tuple[str, ...] = () + artifacts: tuple[dict[str, JsonValue], ...] = () + work_units: tuple[dict[str, JsonValue], ...] = () + implemented_work_ids: tuple[str, ...] = () + work_items: dict[str, WorkItemSnapshot] = Field(default_factory=dict) + + +class ImplementationOutput(DomainModel): + work_unit: dict[str, JsonValue] + context_artifacts: tuple[dict[str, JsonValue], ...] + instructions: str + summary: str | None = None + + +def _current(artifact: dict[str, JsonValue]) -> bool: + status = artifact.get("status") + return status is None or ( + status == "approved" + and bool(artifact.get("digest")) + and artifact.get("approved_digest") == artifact.get("digest") + ) + + +def _issue_artifact(kind: str, item: WorkItemSnapshot, repository: str) -> dict[str, JsonValue]: + content = item.description.strip() + if kind == "task" and not content: + content = item.summary.strip() + return { + "id": f"jira:{item.key}:{kind}", + "kind": kind, + "source": item.key, + "content": content, + "digest": f"sha256:{hashlib.sha256(content.encode()).hexdigest()}", + "repository": repository, + } + + +def run_implementation_input_station( + request: StationRequest[ImplementationInput], +) -> StationOutcome[ImplementationOutput]: + """Select work using only the request payload, with no provider or graph access.""" + data = request.input + repository = data.repository + stale = [ + unit.get("id") + for unit in data.work_units + if unit.get("kind") == "task" + and unit.get("repository") == repository + and unit.get("status") == "stale" + ] + if stale: + raise ValueError(f"Repository {repository} has Tasks derived from stale planning: {stale}") + if data.configured_repository_tasks and not data.candidate_task_keys: + raise NoPendingImplementationWork(f"All Jira tasks are complete for {repository}") + + artifacts: list[dict[str, JsonValue]] = [] + summaries: dict[str, str] = {} + if data.candidate_task_keys: + item = data.work_items[data.candidate_task_keys[0]] + artifact = _issue_artifact("task", item, repository) + if artifact["content"]: + artifacts.append(artifact) + summaries[str(artifact["id"])] = item.summary + + for key in data.epic_keys: + item = data.work_items[key] + repos = {label.removeprefix("repo:") for label in item.labels if label.startswith("repo:")} + if repository not in repos: + continue + artifact = _issue_artifact("epic_plan", item, repository) + if artifact["content"]: + artifacts.append(artifact) + summaries[str(artifact["id"])] = item.summary + + existing_ids = {item.get("id") for item in artifacts} + rank = {"task": 0, "epic_plan": 1, "plan": 2, "spec": 3, "rca": 4, "prd": 5} + for original in sorted(data.artifacts, key=lambda item: rank.get(str(item.get("kind")), 99)): + artifact = dict(original) + if artifact.get("id") in existing_ids or not _current(artifact): + continue + if artifact.get("repository") not in {None, repository} or not artifact.get("content"): + continue + artifacts.append(artifact) + existing_ids.add(artifact.get("id")) + + if data.ticket_key and data.ticket_key not in data.candidate_task_keys[:1]: + item = data.work_items[data.ticket_key] + repos = {label.removeprefix("repo:") for label in item.labels if label.startswith("repo:")} + if repos and repository not in repos: + raise ValueError( + f"Jira issue {item.key} is scoped to {sorted(repos)}, not current repository {repository}" + ) + artifact = _issue_artifact("ticket", item, repository) + if artifact["content"]: + artifacts.append(artifact) + summaries[str(artifact["id"])] = item.summary + + if not artifacts: + raise ValueError(f"No implementation artifact is available for repository {repository}") + primary = artifacts[0] + source = primary.get("source") + jira_key = str(source) if primary.get("kind") in {"task", "epic_plan", "ticket"} else None + digest = str(primary["digest"]) + work_id = jira_key or f"internal:{repository}:{primary['kind']}:{digest[7:19]}" + if work_id in data.implemented_work_ids: + raise NoPendingImplementationWork(f"Work unit {work_id} is already complete") + work_unit: dict[str, JsonValue] = { + "id": work_id, + "kind": primary["kind"], + "key": jira_key, + "repository": repository, + "status": "pending", + "source_artifact_ids": [primary["id"]], + "context_artifact_ids": [item["id"] for item in artifacts[1:]], + } + return StationOutcome[ImplementationOutput]( + workflow=request.workflow, + invocation=request.invocation, + contract_name=request.contract_name, + contract_version=request.contract_version, + status=StationOutcomeStatus.SUCCEEDED, + completed_at=request.requested_at, + output=ImplementationOutput( + work_unit=work_unit, + context_artifacts=tuple(artifacts), + instructions=str(primary["content"]), + summary=summaries.get(str(primary["id"])), + ), + ) diff --git a/src/forge/workflow/stations/runner.py b/src/forge/workflow/stations/runner.py new file mode 100644 index 00000000..66c8b0bc --- /dev/null +++ b/src/forge/workflow/stations/runner.py @@ -0,0 +1,34 @@ +"""Minimal local runner for contract-backed stations.""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +from forge.domain import StationRequest +from forge.workflow.stations.implementation_input import ( + ImplementationInput, + run_implementation_input_station, +) + + +def run_serialized(station_name: str, request_json: str) -> str: + """Run a station from serialized input without the Forge control plane.""" + if station_name != "implementation-input": + raise ValueError(f"Unknown station: {station_name}") + request = StationRequest[ImplementationInput].model_validate_json(request_json) + return run_implementation_input_station(request).model_dump_json() + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("station") + parser.add_argument("request", nargs="?", help="Request JSON file; defaults to stdin") + args = parser.parse_args() + request_json = Path(args.request).read_text() if args.request else sys.stdin.read() + print(run_serialized(args.station, request_json)) + + +if __name__ == "__main__": + main() diff --git a/tests/unit/domain/test_architecture.py b/tests/unit/domain/test_architecture.py index 79333d69..382c186d 100644 --- a/tests/unit/domain/test_architecture.py +++ b/tests/unit/domain/test_architecture.py @@ -4,6 +4,7 @@ from pathlib import Path DOMAIN_ROOT = Path(__file__).parents[3] / "src" / "forge" / "domain" +STATIONS_ROOT = Path(__file__).parents[3] / "src" / "forge" / "workflow" / "stations" PROHIBITED_PREFIXES = ( "langgraph", "redis", @@ -29,3 +30,17 @@ def test_domain_contracts_do_not_import_runtime_or_provider_packages() -> None: violations.append(f"{path.name}:{node.lineno}: {module}") assert not violations, "Prohibited domain dependencies:\n" + "\n".join(violations) + + +def test_stations_do_not_import_providers_or_complete_workflow_state() -> None: + violations: list[str] = [] + for path in sorted(STATIONS_ROOT.glob("*.py")): + tree = ast.parse(path.read_text(), filename=str(path)) + for node in ast.walk(tree): + module = node.module if isinstance(node, ast.ImportFrom) else None + imported = [alias.name for alias in node.names] if isinstance(node, ast.Import) else [] + for name in [*imported, *([module] if module else [])]: + if name.startswith(("forge.integrations", "forge.workflow.base", "langgraph")): + violations.append(f"{path.name}:{node.lineno}: {name}") + + assert not violations, "Prohibited station dependencies:\n" + "\n".join(violations) diff --git a/tests/unit/workflow/test_implementation_input_station.py b/tests/unit/workflow/test_implementation_input_station.py new file mode 100644 index 00000000..a88eda70 --- /dev/null +++ b/tests/unit/workflow/test_implementation_input_station.py @@ -0,0 +1,93 @@ +from datetime import UTC, datetime + +import pytest + +from forge.domain import StationInvocationIdentity, StationRequest, WorkflowIdentity +from forge.workflow.reducers.implementation_input import reduce_implementation_input +from forge.workflow.stations.implementation_input import ( + CONTRACT_NAME, + CONTRACT_VERSION, + ImplementationInput, + run_implementation_input_station, +) +from forge.workflow.stations.runner import run_serialized + +NOW = datetime(2026, 8, 27, tzinfo=UTC) + + +def request() -> StationRequest[ImplementationInput]: + return StationRequest[ImplementationInput]( + workflow=WorkflowIdentity(run_id="run-1", workflow_name="feature", definition_revision=1), + invocation=StationInvocationIdentity( + invocation_id="invocation-1", station_name=CONTRACT_NAME + ), + contract_name=CONTRACT_NAME, + contract_version=CONTRACT_VERSION, + attempt=1, + requested_at=NOW, + input=ImplementationInput( + repository="acme/api", + ticket_key=None, + artifacts=( + { + "id": "plan:1", + "kind": "plan", + "content": "Implement it", + "digest": "sha256:plan", + "approved_digest": "sha256:plan", + "status": "approved", + }, + ), + ), + ) + + +def test_identical_requests_produce_identical_outcomes() -> None: + first = run_implementation_input_station(request()) + second = run_implementation_input_station(request()) + + assert first == second + + +def test_local_runner_round_trips_without_control_plane() -> None: + serialized = run_serialized(CONTRACT_NAME, request().model_dump_json()) + + assert '"status":"succeeded"' in serialized + assert '"instructions":"Implement it"' in serialized + + +def test_reducer_owns_only_documented_checkpoint_fields() -> None: + station_request = request() + outcome = run_implementation_input_station(station_request) + + update = reduce_implementation_input({"unrelated": "preserved"}, station_request, outcome) + + assert set(update) == { + "artifacts", + "work_units", + "current_work_unit_id", + "work_resolution", + } + assert "unrelated" not in update + + +def test_reducer_rejects_stale_invocation() -> None: + station_request = request() + outcome = run_implementation_input_station(station_request).model_copy( + update={ + "invocation": StationInvocationIdentity( + invocation_id="other", station_name=CONTRACT_NAME + ) + } + ) + + with pytest.raises(ValueError, match="does not belong"): + reduce_implementation_input({}, station_request, outcome) + + +def test_reducer_rejects_request_for_another_checkpoint_run() -> None: + station_request = request() + outcome = run_implementation_input_station(station_request) + + with pytest.raises(ValueError, match="checkpoint workflow run"): + reduce_implementation_input({"thread_id": "other-run"}, station_request, outcome) From dcfa71cd577b7d4a36c7cb408073a9d0e357fcfc Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Mon, 31 Aug 2026 10:10:34 +0300 Subject: [PATCH 4/4] ci: run gates for stacked pull requests --- .github/workflows/ci.yml | 5 ----- 1 file changed, 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 77d7f507..2f9ae5e7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -2,11 +2,6 @@ name: CI on: pull_request: - branches: - - main - - integration/dev-to-main - - prototype/declarative-workflows - - prototype/layered-planning-state jobs: helm-lint: