[fas89]

What and why

A team running long-lived Flink SQL jobs (Kafka in, Kafka out) on the Flink Kubernetes Operator and on AWS Managed Service for Apache Flink asked for a way to declare Flink as a build's engine. Their only option was engine: custom, which loses the engine's identity. Worse, it sent the job down the inline-SQL path:

• is_embedded_sql_build never looked at properties.language;
• fluid apply --mode amend-and-build ran the Flink SQL on DuckDB when no platform was declared, and refused platform: kubernetes as an unknown embedded-SQL platform;
• the local, AWS, GCP and Snowflake planners turned it into DuckDB, Athena, BigQuery or Snowflake statements.

This PR:

• Schema (0.7.6 preview only). Adds flink to the build engine enum. 0.7.5 is unchanged. Deployment settings go under properties.parameters, because embeddedLogicPattern is closed. The schema description says so and names the Flink Kubernetes Operator's job spec keys (parallelism, upgradeMode) and flinkConfiguration.
• One predicate, util/contract.is_externally_run_build. It matches engine: flink, and also properties.language: flink_sql, which is how a 0.7.5 contract declares the same job.
  • Every consumer that ran or planned a build's SQL now checks it: the build runner, the local planner and provider, and the AWS, GCP and Snowflake planners.
  • Snowflake's replace mode still ensures a Flink output table, since no CTAS recreates it.
  • Other properties.language values keep the behaviour they had.
• Runner output. fluid apply --mode amend-and-build reports such a build as run outside forge-cli and counts it on its own summary line, not as a skip. So a contract whose only builds run elsewhere is not "every build was skipped". The run report records it as skipped with reason: runs_outside_forge.
• Generators.
  • Scheduler synthesis and the Airflow fluid-apply DAGs give these builds no task.
  • fluid generate schedule on a contract with no other build, and no hand-declared tasks, exits 0 with "nothing to schedule". An unknown --scheduler is still an error.
  • fluid generate transformation says there is no code to generate.
  • The local planner lands nothing for an expose such a build writes. When a contract mixes such a build with a local one and builds[].outputs does not say which writes which expose, it lands neither and warns rather than guess. The local plan of all 37 repo contracts is unchanged from main.
• No runner. As the requester proposed, forge-cli validates and plans a flink build; the platform that runs the job executes it.
• fix(local) (its own commit in this PR). The local planner took the output table of the contract's last build. When that build produced no table (a dbt model), it landed the first loaded input raw in the expose, columns the expose does not declare included. It now takes the last build that produced a table.

The requester asked about trigger for a continuous job: trigger is already optional in the schema, so a continuous job omits it. A continuous trigger type is left until a scheduler needs to tell always-on from manual.

Borrowed, not built

• Flink Kubernetes Operator custom-resource reference. JobSpec parallelism and upgradeMode (savepoint / last-state / stateless), and FlinkDeploymentSpec flinkConfiguration. These are the recommended properties.parameters names; there is no new typed block, which RFC-streaming-extension §12 also rejected.
• OpenLineage JobTypeJobFacet. Integration FLINK, processingType STREAMING; it confirms "flink" as the engine identity.
• No runner exists to borrow. forge-cli has no build-runner plugin hook (its plugin groups are providers, validators, catalogs, scaffolds, iac, commands, apply hooks and schemas), so execution stays with the job's platform.

Validation

• Tests:
  • tests/test_flink_engine.py (55 tests) and tests/providers/test_local_planner_source_table.py.
  • All 10 original behaviour tests fail on main.
  • The round-1 mutants were all killed: ignore language, the Snowflake expose guard, the report reason, the synthesis filter, the AWS guard, count external builds as skipped.
• Full suite:
  • 26,622 passed. The 6 failures are environmental and reproduce on unmodified main with the same venv: duckdb 1.5.6 sandbox tests, dbt not installed, the live coding-agent test.
  • Perf startup budget: 38 passed.
  • Lint, format and dead code: whole-tree ruff, black 24.10.0 (the CI pin), the dead-code gate, check_license_headers.py and lint-imports.
• Live through the fluid CLI, on the requester's contract (parallelism moved under properties.parameters) and on its 0.7.5 engine: sql + language: flink_sql twin:
  • validate passes on 0.7.6 and rejects flink on 0.7.5.
  • plan passes.
  • apply --mode amend-and-build, both dry run and real, prints "engine 'flink' runs on kubernetes, outside forge-cli", "Run outside forge-cli: 1", exit 0. On main the dry run printed "Would execute SQL: INSERT INTO …"; on 0.7.5 main exited 1 with embedded_sql_platform_unsupported.
  • generate transformation and generate schedule exit 0 with their notes.
• ripwire --quality-delta, measured against a no-op control on main. Accepted rows:
  • run_builds_from_args grows 96 → 99, the minimum one dispatch branch plus a summary line costs inside an already 96-complexity function;
  • the local planner's plan_actions grows by a skip;
  • a false dead-code row on ApplyRunReport.record_build, which is called through a dynamically typed report;
  • a test-helper duplication.

Review verdicts

• security-review (three rounds): no findings. The skips only stop SQL or DAG emission, and no masking, governance or sovereignty step sits behind them. A sub-threshold note from round 2 (a local expose landing raw input when every build runs elsewhere) is fixed here.
• world-class-inspection. Round 1: architecture PASS; documentation, codebase and test suite CONCERNS. Every change-caused finding is fixed and re-verified:
  • the language case;
  • the upgrade-note wording;
  • the flinkConfiguration naming;
  • schedule and transformation generation;
  • the report reason;
  • tests over both shapes;
  • the Snowflake replace-mode test.
• defect-scan (two hunters, three re-verify rounds). All change-caused findings are fixed and re-verified, the last being a mixed contract that landed a SQL build's table under a Flink expose's name. Pre-existing ones are follow-ups below.

Documentation

• Docs PR linked: docs: engine: flink, and overlays that merge by identity from 0.7.6 (forge-cli #731, #732) forge_docs#155 (Flink jobs section, identity-merge overlay rules; merge after this PR)
• No docs needed
• Docs TODO

Follow-ups (not in this PR)

• fluid apply (simple mode) picks a build's runtime.platform (kubernetes) as the provider: provider_unknown, the same root cause as provider_unknown kafka.
• The copilot doesn't know flink yet (KNOWN_BUILD_ENGINES feeds its prompt, and it writes 0.7.5 contracts).
• policy.privacy.masking on an expose a Flink or Python build writes gets no advisory.
• Other properties.language values (pyspark, python, scala, r) with properties.sql still run on DuckDB, and forge's Python templates report a false success.
• The local planner's fallback still lands the first loaded input when no build produced a table and no expose id matches (also when a Flink build declares outputs naming another expose).
• fluid plan's abstract scheduleTask still carries the Flink SQL (nothing executes it).

[github-actions]

📄 Documentation Reminder

This PR appears to be missing a documentation reference. Our docs live in a separate repo.

Please update the PR description with one of:

• Link a docs PR — check the "Docs PR linked" box and paste the URL
• Mark as no docs needed — check "No docs needed" with a justification
• Acknowledge docs TODO — check "Docs TODO" and create the docs PR before merge

See the Contributing Guide for details.