Add Go-SDK Agent Skills

skills/authoring-go-sdk-tasks/SKILL.md

@@ -0,0 +1,173 @@
+---
+name: authoring-go-sdk-tasks
+description: Writes Airflow task logic in Go using the Airflow Go SDK. Use when the user wants to implement Airflow tasks in Go, asks about `BundleProvider`/`RegisterDags`, the `bundlev1` Registry/Dag interfaces, registering Go tasks (`AddTask`/`AddTaskWithName`), dependency injection by parameter type (`context.Context`, `sdk.TIRunContext`, `*slog.Logger`, `sdk.Client`), or reading connections/variables/XComs from Go. This skill covers the Go-specific native API; the shared Python-stub pattern and conceptual model live in authoring-language-sdk-tasks. For building/packing/shipping the bundle see deploying-go-sdk-bundles; for coordinator config see configuring-airflow-language-sdks.
+---
+
+# Authoring Go SDK Tasks
+
+The Airflow Go SDK implements the language-SDK model for Go: your DAG stays in Python, and each task is a compiled Go function registered inside a **bundle** (a single native executable). This skill covers the **Go-specific** native API. The shared model (the Python `@task.stub` pattern, ID matching, the XCom-as-JSON contract) lives in **authoring-language-sdk-tasks**; read that first if you are new to language SDKs.
+
+> **Experimental.** The Go SDK is under active development and not production-ready. Module path `github.com/apache/airflow/go-sdk` (Go 1.24+). APIs may change.
+
+> **Related skills:** **authoring-language-sdk-tasks** (shared Python stub + concepts), **deploying-go-sdk-bundles** (build, pack, and ship the bundle), **configuring-airflow-language-sdks** (route the queue to the Go coordinator).
+
+---
+
+## Recap: the Python side
+
+A Go task is paired with a Python stub that carries no logic; it declares the task, its queue, and the dependency graph. IDs must match the Go registration exactly, and `queue=` routes the task to the Go runtime. Full rules are in **authoring-language-sdk-tasks**; the minimal shape:
+
+```python
+from airflow.sdk import dag, task
+
+
+@task.stub(queue="golang")
+def extract(): ...
+
+
+@task.stub(queue="golang")
+def transform(): ...
+
+
+@dag()
+def simple_dag():
+    extract() >> transform()
+
+
+simple_dag()
+```
+
+The `queue` value (`"golang"` here) is an arbitrary label that must match the queue routed to the Go coordinator (`queue_to_coordinator`). See **configuring-airflow-language-sdks**.
+
+---
+
+## The bundle entry point
+
+A bundle implements `bundlev1.BundleProvider`: report its version and register your DAGs and tasks. `main` is one line; `bundlev1server.Serve` wires the bundle to the Airflow runtime for you.
+
+```go
+package main
+
+import (
+	"log"
+
+	v1 "github.com/apache/airflow/go-sdk/bundle/bundlev1"
+	"github.com/apache/airflow/go-sdk/bundle/bundlev1/bundlev1server"
+)
+
+type myBundle struct{}
+
+var _ v1.BundleProvider = (*myBundle)(nil)
+
+func (m *myBundle) GetBundleVersion() v1.BundleInfo {
+	return v1.BundleInfo{Name: bundleName, Version: &bundleVersion}
+}
+
+func (m *myBundle) RegisterDags(dagbag v1.Registry) error {
+	simpleDag := dagbag.AddDag("simple_dag")      // dag_id must match the Python @dag name
+	simpleDag.AddTask(extract)                    // task_id is the function name; must match the stub
+	simpleDag.AddTaskWithName("transform", transform) // or set the task_id explicitly
+	return nil
+}
+
+func main() {
+	if err := bundlev1server.Serve(&myBundle{}); err != nil {
+		log.Fatal(err)
+	}
+}
+```
+
+`AddTask(fn)` derives the `task_id` from the Go function's name; use `AddTaskWithName("<task_id>", fn)` when that name can't match the Python stub (an unexported, renamed, or reused function). `RegisterDags` is the single source of truth for task identity: the bundle's manifest (used by the packer and by the coordinator) is generated by running it, never hand-written.
+
+---
+
+## Task functions: dependency injection by parameter type
+
+A task is an ordinary Go function. The runtime inspects its signature and injects arguments **by type**; declare only what you need.
+
+| Parameter type | Injected value |
+|----------------|----------------|
+| `context.Context` | Task context for cancellation. Always available. |
+| `sdk.TIRunContext` | Richer context (embeds `context.Context`) exposing `TaskInstance()` and `DagRun()`. See [Runtime context](#runtime-context). |
+| `*slog.Logger` | Logger wired to the Airflow task log. |
+| `sdk.Client` | Full Airflow model access: Variables, Connections, XComs. |
+| `sdk.VariableClient` / `sdk.ConnectionClient` / `sdk.XComClient` | A narrower slice of `sdk.Client`. Prefer the narrowest you need; it documents intent and is trivial to fake in tests. |
+
+The optional return signature is `(result, error)`: a non-nil `result` is pushed as the task's `return_value` XCom; a non-nil `error` fails the task (which triggers the stub's retry policy). Returning only `error`, or nothing, is also valid.
+
+```go
+func extract(ctx sdk.TIRunContext, client sdk.Client, log *slog.Logger) (any, error) {
+	conn, err := client.GetConnection(ctx, "test_http")
+	if err != nil {
+		return nil, err
+	}
+	log.Info("connected", "host", conn.Host)
+	return map[string]any{"go_version": runtime.Version()}, nil
+}
+
+func transform(ctx sdk.TIRunContext, client sdk.VariableClient) error {
+	val, err := client.GetVariable(ctx, "my_variable")
+	if err != nil {
+		return err // VariableNotFound (a sentinel error) if absent
+	}
+	_ = val
+	return nil
+}
+```
+
+---
+
+## The `sdk.Client` surface
+
+| Call | Returns | Notes |
+|------|---------|-------|
+| `GetVariable(ctx, key)` | `(string, error)` | `VariableNotFound` if absent. |
+| `UnmarshalJSONVariable(ctx, key, &ptr)` | `error` | Decode a JSON variable into a struct/pointer. |
+| `GetConnection(ctx, connID)` | `(Connection, error)` | `ConnectionNotFound` if absent. |
+| `GetXCom(ctx, dagID, runID, taskID, mapIndex, key, value)` | `(any, error)` | `XComNotFound` only if the key is absent; a stored null returns `(nil, nil)`. |
+| `PushXCom(ctx, ti, key, value)` | `error` | Rarely needed; a returned value is pushed for you. |
+
+`Connection` exposes `ID`, `Type`, `Host`, `Port` (`int`), `Login *string`, `Password *string` (nil when unset, distinct from empty), `Path` (schema), `Extra map[string]any`, plus `GetURI()`. Not-found cases return the sentinels `sdk.VariableNotFound`, `sdk.ConnectionNotFound`, `sdk.XComNotFound`.
+
+To read an upstream task's result, call `GetXCom` explicitly, taking the `dag_id`/`run_id`/`task_id` you need from the runtime context (below).
+
+---
+
+## Runtime context
+
+Declare an `sdk.TIRunContext` parameter to read metadata about the task instance and its DAG run. It is an interface that embeds `context.Context`, so it is usable anywhere a `context.Context` is expected.
+
+```go
+func extract(ctx sdk.TIRunContext, log *slog.Logger) error {
+	ti, dagRun := ctx.TaskInstance(), ctx.DagRun()
+	log.Info("running",
+		"task_id", ti.TaskID,
+		"run_id", dagRun.RunID,
+		"logical_date", dagRun.LogicalDate)
+	return nil
+}
+```
+
+- `TaskInstance()`: `DagID`, `RunID`, `TaskID`, `MapIndex *int` (nil when unmapped), `TryNumber`.
+- `DagRun()`: `DagID`, `RunID`, and the `*time.Time` timestamps `LogicalDate`, `DataIntervalStart`, `DataIntervalEnd` (nil when not sent).
+
+The accessors are populated from the task's startup details before the body runs. Because `TIRunContext` embeds `context.Context`, pass it straight to client calls and cancellation checks (`ctx.Done()`); declare it as your context parameter by default. In tests, build the argument with `sdk.NewTIRunContext(ctx, ti, dagRun)` (it panics on a nil `ctx`).
+
+---
+
+## Go-specific pitfalls
+
+- **IDs must match the Python stub** (`dag_id` from `AddDag`, `task_id` from the registered function name), and the stub's `queue=` must route to the Go coordinator, or the task is never delivered.
+- **`RegisterDags` is authoritative.** Do not hand-write the manifest; the packer generates it by running `RegisterDags`.
+- **Ask for the narrowest client interface** you need (`sdk.VariableClient` over `sdk.Client`) for clearer intent and easier fakes.
+- **A non-nil `error` return fails the task** and applies the stub's retries; a recovered panic is also a failure.
+- See **authoring-language-sdk-tasks** for the language-agnostic pitfalls (one process per task instance, set queue and retries on the stub).
+
+---
+
+## Related Skills
+
+- **authoring-language-sdk-tasks**: Shared Python-stub pattern and concepts (read first).
+- **deploying-go-sdk-bundles**: Build and pack the bundle with `go tool airflow-go-pack`, then deploy it for the coordinator.
+- **configuring-airflow-language-sdks**: Route the queue to the Go coordinator (`ExecutableCoordinator`).
+- **authoring-dags**: General Airflow DAG authoring.

skills/authoring-language-sdk-tasks/SKILL.md

@@ -0,0 +1,123 @@
+---
+name: authoring-language-sdk-tasks
+description: The language-neutral foundation for Airflow language SDKs — implement task logic in a non-Python language while the DAG stays in Python. Use when the user wants to run an Airflow task in another language (Go or other native languages), asks how the Python `@task.stub` pairs with native task code, how task/DAG IDs must match across the two sides, how data passes via XCom as JSON, or which language SDKs exist. This skill owns the shared Python-stub pattern and conceptual model; for a specific language's native API, build, and runtime, use that language's skill (e.g. authoring-go-sdk-tasks).
+---
+
+# Authoring Language SDK Tasks (Shared Foundation)
+
+Airflow language SDKs let you implement task logic in a language other than Python while the DAG and its scheduling stay in Python. This skill describes the parts that are identical across every language SDK. Each language has its own companion skill for the native API, build tooling, and runtime — see [Per-language skills](#per-language-skills).
+
+> **Experimental.** The language SDKs are in preview. APIs and artifact coordinates may change.
+
+---
+
+## The model
+
+A DAG is authored in Python as usual. Tasks that should run in another language are declared as **stubs** routed to a dedicated queue. At runtime, Airflow hands a stub task to a **coordinator** that launches a short-lived **native subprocess** for that one task instance, runs your compiled/native code, and shuts the subprocess down.
+
+Consequences that hold for every language SDK:
+
+- **One subprocess per task instance** — there is no shared in-process state between task instances. Pass data via XCom or an external store.
+- **The DAG, schedule, retries, and queue routing live in Python.** The native side only implements task logic.
+- **Data crossing the boundary is JSON.** See [The XCom-as-JSON contract](#the-xcom-as-json-contract).
+
+---
+
+## The two-sided model
+
+Every task has two halves that must agree:
+
+1. A **Python stub** in a normal DAG file — no logic; it declares the task, its queue, the dependency graph, and retry policy.
+2. A **native implementation** (Go, etc.) whose IDs match the Python side and where the work happens.
+
+### Python side (scheduling)
+
+The example below uses the Go SDK to be concrete, but the Python side is **identical for every language SDK**. The queue name (`"golang"` here) is an arbitrary label you choose — it just has to match a key in `queue_to_coordinator` (see **configuring-airflow-language-sdks**). Pick whatever name fits the SDK you're routing to.
+
+```python
+from datetime import timedelta
+from airflow.sdk import dag, task
+
+
+@dag
+def sales_pipeline():
+    @task.stub(queue="golang")          # queue selects the coordinator (see configuring-airflow-language-sdks)
+    def extract(): ...
+
+    @task.stub(queue="golang")
+    def transform(extracted): ...        # arg only declares the dependency
+
+    @task.stub(queue="golang", retries=1, retry_delay=timedelta(seconds=5))
+    def load(transformed): ...
+
+    @task()                              # an ordinary Python task can sit downstream
+    def report(loaded):
+        print(f"done: {loaded}")
+
+    report(load(transform(extract())))
+
+
+sales_pipeline()
+```
+
+Rules that apply regardless of language:
+
+- The **stub function name is the task ID** and the `@dag` name (or `dag_id=`) is the DAG ID. The native side must use these exact IDs.
+- An upstream argument on a stub (e.g. `transform(extracted)`) exists **only to declare the dependency** in Python. The value itself is fetched on the native side via XCom — passing it in Python does not hand it to the native code.
+- **Queue, retries, and other task arguments are set on the stub**, not in the native code. A native task that fails is reported back to Airflow, which then applies the stub's retry policy.
+- The `queue` value is what routes the task to a coordinator; the same string must appear in `queue_to_coordinator` (see **configuring-airflow-language-sdks**).
+
+---
+
+## The XCom-as-JSON contract
+
+XCom values are stored as JSON in Airflow's metadata database, so the boundary between Python and any native language is JSON. The Python/JSON side is the same for every SDK:
+
+| Python type | JSON |
+|-------------|------|
+| `int` | number (integer) |
+| `float` | number (decimal) |
+| `str` | string |
+| `bool` | boolean |
+| `None` | null |
+| `list` | array |
+| `dict` | object |
+
+Each language SDK maps these JSON types onto its own native types. The native-type mapping lives in that language's skill. The key portability rule: a value pushed by one task is read by another **as JSON**, so the consuming side must expect a type compatible with what was stored.
+
+---
+
+## What is language-specific (and lives elsewhere)
+
+This skill deliberately stops at the shared concepts. The following differ per language and are documented in each language's companion skills:
+
+- **Native task API** — how you declare tasks, read connections/variables/XComs, and push results (annotations, interfaces, function registration, etc.).
+- **Native type mapping** — the native column of the JSON table above.
+- **Build and packaging** — how the artifact is compiled and bundled.
+- **Runtime prerequisite** — what must be present on the worker (a language runtime for some SDKs; none for the Go SDK's self-contained bundles).
+
+The Airflow-side wiring (which coordinator runs which queue) is shared in structure but has per-coordinator options; it lives in **configuring-airflow-language-sdks**.
+
+---
+
+## Language-agnostic pitfalls
+
+- **IDs must match exactly** across the Python stub function name and the native task ID, and across `@dag`/`dag_id` and the native DAG ID. Mismatches surface as "no DAGs" or missing-XCom errors.
+- **Both sides need the upstream reference.** Python declares the dependency by passing the upstream call; the native code retrieves the value via XCom.
+- **Set queue and retries on the stub**, never in the native code.
+- **Stub bodies must be empty.** An AST check enforces it — only `pass`, `...`, or a docstring is allowed in the body; any real logic is rejected.
+- **`retry_policy` is rejected on stubs** (`@task.stub` raises `ValueError`). Use `retries`/`retry_delay` instead — a retry-policy callable runs Python in-process and would never fire for a task executing in a native subprocess.
+- Assets, deferral, and some other Airflow features have limited or no support in the language SDKs today.
+
+---
+
+## Per-language skills
+
+- **authoring-go-sdk-tasks**: Go native API — task registration, dependency injection by parameter type, and client access.
+- *(Future language SDKs each add their own `authoring-<lang>-sdk-tasks` skill that builds on this one.)*
+
+## Related Skills
+
+- **configuring-airflow-language-sdks**: Route a queue to a coordinator and set runtime options.
+- **authoring-dags**: General Airflow DAG authoring (the Python side lives here too).
+- **deploying-go-sdk-bundles**: Build, pack, and ship the Go bundle (per-language deploy skills follow the same shape).