Add skills for AI agents

.agents/skills/fkafka/SKILL.md

@@ -0,0 +1,66 @@
+---
+name: fkafka
+description: Use whenever generating or reviewing F# code that produces to or consumes from Apache Kafka via the Alma.Kafka library — calls to Consumer.consume, ConsumerConfiguration.createWithConnection, Producer.create / Producer.produce, MessageToProduce, Admin.lags, Checker, or composes TracedMessage handling, manual/auto commit (CommitMessage), external offset checkpoints (GetCheckpoint), or B3 trace propagation through Kafka headers. Trigger also on mentions of BrokerList, StreamName, GroupId, consumer lag, KSQL message keys, MessageKey.Delimited, or "read/write a Kafka topic in F#".
+---
+
+# F-Kafka
+
+Library: [alma-oss/fkafka](https://github.com/alma-oss/fkafka)
+NuGet: `Alma.Kafka`
+
+## Purpose
+
+`Alma.Kafka` is an F# wrapper over `Confluent.Kafka` for producing and consuming messages to/from Kafka topics. It exposes a typed, railway-oriented (`AsyncResult`) API with consumer-lag monitoring, cluster/topic health checking, manual or automatic offset commit, external offset checkpoints, and automatic distributed-trace propagation through Kafka message headers.
+
+## When to Use
+
+- Consuming a Kafka topic as a lazy F# sequence of typed messages.
+- Producing messages with a key (simple or KSQL-style delimited) and custom headers.
+- Monitoring consumer lag per partition for a given group.
+- Adding health checks (cluster/topic availability with retry) around producing/consuming.
+- Storing offsets in external storage instead of Kafka's built-in commit.
+
+## When NOT to Use
+
+- Non-Kafka message brokers.
+- Low-level partition assignment or admin operations beyond topic listing and lag — drop to `Confluent.Kafka` directly.
+- Schema-registry / Avro / Protobuf serialization — this library works with `string` payloads.
+
+## Main Concepts
+
+- `BrokerList` — single-case DU wrapping the comma-separated bootstrap-server string.
+- `StreamName` — topic name; either an explicit `StreamName` or derived from an `Instance`.
+- `GroupId` — `Id of string` for a stable group, or `Random` for a unique throwaway group (always reads from the beginning).
+- `ConnectionConfiguration` — record of `{ BrokerList; Topic }`.
+- `ConsumerConfiguration` — full consumer setup; build with `ConsumerConfiguration.createWithConnection` or `createWithDefaults`.
+- `Consumer.consume` — turns a configuration into a lazy, effectively infinite `seq` of `TracedMessage`.
+- `TracedMessage<'Message>` — `{ Commit; Message; Trace }`; map the payload via `TracedMessage.map`, finish the span via `TracedMessage.finish`.
+- `CommitMessage` — `Automatically` (Kafka autocommit) or `Manually of FailOnNotCommittedMessage`.
+- `ManualCommit` — handle whose `ManualCommit.execute` commits the current offset under manual mode.
+- `GetCheckpoint` — optional function to resolve a starting offset per partition from external storage.
+- `ProducerConfiguration` / `Producer` — producer setup and the disposable producer handle.
+- `MessageToProduce` — `{ Key; Headers; Value }`; build with `MessageToProduce.create` / `createWithHeaders`.
+- `MessageKey` — `Simple of string` or `Delimited of string list` (joined with `,`, spaces stripped, KSQL-compatible).
+- `Admin` — cluster inspection: `createAdmin`, `getAllTopics`, `topicExists`, `isUp`, `lags`.
+- `PartitionLag` — `{ Partition; Lag }` produced by `Admin.lags`.
+- `Checker` / `IntervalChecker` — health-check records with `defaultChecker` presets.
+- `Event<'KeyData,'MetaData,'DomainData>` / `CommonEvent` — typed event-envelope schema with `EventId`, `CorrelationId`, `CausationId`, `Resource`.
+- `MetaData` — parsed message metadata (`OnlyCreatedAt` or `CreatedAndProcessed`) via `MetaData.parse`.
+
+## Related Libraries
+
+- `Confluent.Kafka` — underlying client; its types surface in errors and handles.
+- `Feather.ErrorHandling` — `AsyncResult` / `asyncResult` CE used across the API.
+- `Alma.Tracing` — span types and B3 header inject/extract used for trace propagation.
+- `Alma.ServiceIdentification` — `Instance`, `Domain`, `Context` used by `StreamName` and event envelopes.
+- `Alma.Metrics` — service status (`MarkAsEnabled` / `MarkAsDisabled`) wired into health checks.
+
+## Keywords for Search
+
+Kafka, Alma.Kafka, fkafka, F# Kafka, Confluent.Kafka, consumer, producer, BrokerList, StreamName, GroupId, ConsumerConfiguration, Consumer.consume, TracedMessage, CommitMessage, ManualCommit, GetCheckpoint, checkpoint, offset, consumer lag, Admin.lags, PartitionLag, Producer, MessageToProduce, MessageKey, Delimited, KSQL, Header, Checker, IntervalChecker, health check, trace propagation, B3 headers, Event, MetaData
+
+## Reference Files
+
+- For composition principles and recommended API usage, read `references/preferred-patterns.md`.
+- For known pitfalls and incorrect assumptions, read `references/anti-patterns.md`.
+- For worked code examples, read `references/examples.md`.

.agents/skills/fkafka/references/anti-patterns.md

@@ -0,0 +1,55 @@
+# Anti-Patterns — Alma.Kafka
+
+Each entry is **mistake → why → fix**.
+
+## Lifecycle
+
+- **Mistake:** Creating a `Producer` or `Consumer` without `use` (or never calling `Producer.close`).
+  **Why:** Both hold native `librdkafka` resources and implement `IDisposable`; a leaked producer may never flush buffered messages, and a leaked consumer never closes its group session.
+  **Fix:** Bind with `use producer = Producer.create config`, or scope the `Consumer.consume` sequence so disposal happens when the scope ends.
+
+- **Mistake:** Forcing the consume sequence with `Seq.toList` / `List.ofSeq`.
+  **Why:** `Consumer.consume` is effectively infinite; materializing it blocks forever.
+  **Fix:** Stream it lazily with `Seq.iter` / `Seq.map` / `Seq.truncate`.
+
+## Group IDs
+
+- **Mistake:** Using `GroupId.Random` in production.
+  **Why:** It generates a unique group id per process start, so the consumer always re-reads the topic from the beginning and never shares progress across instances.
+  **Fix:** Use `GroupId.Id "<stable-name>"` for any long-running consumer; reserve `GroupId.Random` for tests and one-off reads.
+
+## Commit Handling
+
+- **Mistake:** Selecting `CommitMessage.Manually` but never calling `ManualCommit.execute`.
+  **Why:** Offsets are never committed, so the consumer reprocesses from the last committed position on restart; with `FailOnNotCommittedMessage.WithException` the next consume fails with `ConsumeError.PreviousMessageWasNotCommited`.
+  **Fix:** After successfully processing a message, call `ManualCommit.execute message.Commit` and handle the `ManualCommitError` result.
+
+- **Mistake:** Discarding the `Result` from `ManualCommit.execute`.
+  **Why:** A failed commit (`ManualCommitError.KafkaException` / `RuntimeException`) is silently lost, masking duplicate-processing risk.
+  **Fix:** Pattern-match the result and log/propagate the error.
+
+## External Checkpoints
+
+- **Mistake:** Throwing from a `GetCheckpoint` function when no stored offset exists.
+  **Why:** The library treats a successful result with `Offset = None` as "no checkpoint" and falls back to the earliest offset; an exception instead aborts partition assignment.
+  **Fix:** Return `Ok { TopicPartition = tp; Offset = None }` when nothing is stored.
+
+- **Mistake:** Saving the external checkpoint and committing the Kafka offset as independent, non-atomic steps.
+  **Why:** A crash between the two leaves stored and committed offsets out of sync, causing message loss or duplication.
+  **Fix:** Persist the external checkpoint and the Kafka commit together (transactionally or with idempotent processing) after each message or batch.
+
+## Message Keys
+
+- **Mistake:** Embedding spaces or your own separators in a `MessageKey.Delimited` list.
+  **Why:** Delimited keys are joined with `,` and have spaces stripped to stay KSQL-compatible; manual separators or spaces produce keys that don't match downstream consumers.
+  **Fix:** Pass the raw field values as a `string list` and let `MessageKey.Delimited` build the comma-joined key.
+
+## Wrong Abstractions
+
+- **Mistake:** Reaching into the `Trace` module or hand-injecting B3 headers.
+  **Why:** Trace inject/extract is internal and runs automatically inside `Producer.produce` and `Consumer.consume`; duplicating it produces conflicting spans.
+  **Fix:** Rely on the built-in propagation; only create application-level child spans through `Alma.Tracing`.
+
+- **Mistake:** Pattern-matching wrapper DUs (`BrokerList`, `Offset`, `GroupId`) inline across modules.
+  **Why:** It bypasses the companion-module API and breaks if the representation changes.
+  **Fix:** Use the module accessors (`BrokerList.value`, `GroupId.value`, …).

.agents/skills/fkafka/references/examples.md

@@ -0,0 +1,171 @@
+# Examples — Alma.Kafka
+
+This file is the single source of truth for all example code in this skill. Each example is self-contained and ordered by increasing complexity. Payloads are `string`; `parseMessage` is a placeholder for your own deserialization.
+
+## Basic Consume
+
+```fsharp
+open Alma.Kafka
+
+let connection = {
+    BrokerList = BrokerList "127.0.0.1:9092"
+    Topic = StreamName "demo-topic"
+}
+
+let configuration = ConsumerConfiguration.createWithConnection connection (GroupId.Id "demo-group")
+
+let parseMessage (raw: string) = raw  // replace with real deserialization
+
+Consumer.consume configuration (TracedMessage.message >> parseMessage)
+|> Seq.iter (fun message -> printfn "Message: %s" message)
+```
+
+## Produce With Key And Headers
+
+```fsharp
+open Alma.Kafka
+
+let connection = {
+    BrokerList = BrokerList "127.0.0.1:9092"
+    Topic = StreamName "demo-topic"
+}
+
+use producer = Producer.create (ProducerConfiguration.createWithConnection connection)
+
+let headers = [
+    Header.ofString (HeaderKey "source") "web-api"
+]
+
+// Simple key
+MessageToProduce.createWithHeaders headers (MessageKey.Simple "entity-42", "payload")
+|> Producer.produce producer
+
+// Composite, KSQL-compatible key (joined with ",")
+MessageToProduce.create (MessageKey.Delimited [ "tenant-1"; "entity-42" ], "payload")
+|> Producer.produceSingle producer   // produce + flush
+```
+
+## Consume With Manual Commit
+
+```fsharp
+open Alma.Kafka
+
+let connection = {
+    BrokerList = BrokerList "127.0.0.1:9092"
+    Topic = StreamName "demo-topic"
+}
+
+let configuration =
+    { ConsumerConfiguration.createWithConnection connection (GroupId.Id "demo-group") with
+        CommitMessage = CommitMessage.Manually FailOnNotCommittedMessage.WithException
+    }
+
+Consumer.consume configuration id
+|> Seq.iter (fun tracedMessage ->
+    // process tracedMessage.Message ...
+    match ManualCommit.execute tracedMessage.Commit with
+    | Ok () -> ()
+    | Error (ManualCommitError.KafkaException e) -> eprintfn "commit failed: %A" e
+    | Error (ManualCommitError.RuntimeException e) -> eprintfn "commit failed: %A" e
+)
+```
+
+## Consumer Lag
+
+```fsharp
+open Microsoft.Extensions.Logging
+open Alma.Kafka
+open Alma.Kafka.Admin
+
+let runLag (logger: ILogger) =
+    let connection = {
+        BrokerList = BrokerList "127.0.0.1:9092"
+        Topic = StreamName "demo-topic"
+    }
+
+    let totalLag =
+        Admin.lags logger connection (GroupId.Id "demo-group")
+        |> Async.RunSynchronously
+        |> List.sumBy PartitionLag.lag
+
+    printfn "total lag: %d" totalLag
+```
+
+## External Checkpoint
+
+```fsharp
+open Alma.Kafka
+open Feather.ErrorHandling
+
+// Resolve a starting offset from external storage; return Offset = None when nothing is stored.
+let getCheckpoint (groupId: GroupId) (topicPartition: TopicPartition): AsyncResult<TopicPartitionOffset, exn> = asyncResult {
+    let! storedOffset = ExternalStore.tryGetOffset groupId topicPartition  // your code: returns Offset option
+    return { TopicPartition = topicPartition; Offset = storedOffset }
+}
+
+let connection = {
+    BrokerList = BrokerList "127.0.0.1:9092"
+    Topic = StreamName "demo-topic"
+}
+
+let configuration =
+    { ConsumerConfiguration.createWithConnection connection (GroupId.Id "demo-group") with
+        CommitMessage = CommitMessage.Manually FailOnNotCommittedMessage.WithException
+        GetCheckpoint = Some getCheckpoint
+    }
+
+Consumer.consume configuration id
+|> Seq.iter (fun tracedMessage ->
+    // process tracedMessage.Message, then persist the external checkpoint and commit atomically
+    match ManualCommit.execute tracedMessage.Commit with
+    | Ok () -> ExternalStore.saveOffset (GroupId.Id "demo-group") tracedMessage  // your code
+    | Error e -> eprintfn "commit failed: %A" e
+)
+```
+
+## Consume With Application Tracing
+
+```fsharp
+open Alma.Kafka
+open Alma.Tracing
+
+let connection = {
+    BrokerList = BrokerList "127.0.0.1:9092"
+    Topic = StreamName "demo-topic"
+}
+
+let configuration = ConsumerConfiguration.createWithConnection connection (GroupId.Id "demo-group")
+
+// The consumer's own span is created automatically from the message's B3 headers.
+// Here we start an application child span for processing and finish it ourselves.
+Consumer.consume configuration (fun tracedMessage ->
+    tracedMessage.Message,
+    "Process message" |> Trace.ChildOf.start tracedMessage.Trace
+)
+|> Seq.iter (fun (message, processTrace) ->
+    // process message ...
+    processTrace |> Trace.finish
+)
+```
+
+## Health Checked Producer
+
+```fsharp
+open Alma.Kafka
+
+let connection = {
+    BrokerList = BrokerList "127.0.0.1:9092"
+    Topic = StreamName "demo-topic"
+}
+
+// Blocks with retry until the cluster and topic are available, then yields a connected producer.
+let configuration =
+    { ProducerConfiguration.createWithConnection connection with
+        Checker = Some Checker.defaultChecker
+    }
+
+use producer = Producer.create configuration
+
+MessageToProduce.create (MessageKey.Simple "entity-42", "payload")
+|> Producer.produceSingle producer
+```

.agents/skills/fkafka/references/preferred-patterns.md

@@ -0,0 +1,42 @@
+# Preferred Patterns — Alma.Kafka
+
+## Core Principles
+
+- Build configurations through the smart constructors (`ConsumerConfiguration.createWithConnection`, `ProducerConfiguration.createWithConnection`) and override individual fields with record-update syntax rather than constructing the records by hand. New optional fields get sane defaults this way.
+- Treat every wrapper type through its companion module (`BrokerList.value`, `StreamName.value`, `GroupId.value`, `PartitionLag.lag`). Do not pattern-match the DU inline outside the module that owns it.
+- Both `Producer` and `Consumer` own native resources and implement `IDisposable`. Bind them with `use` (or `Producer.close` / consume the sequence inside a scope) so the producer flushes and the consumer closes deterministically.
+- `Consumer.consume` returns a lazy, effectively infinite sequence. Drive it with `Seq.iter` / `Seq.map` and let the surrounding scope control lifetime; do not force it with `Seq.toList`.
+
+## Recommended API Usage
+
+- Consuming: pass a projection from `TracedMessage<string>` to your payload as the second argument of `Consumer.consume`. The simplest projection is `TracedMessage.message >> parseMessage`. See `examples.md` → Basic Consume.
+- Producing: build the value with `MessageToProduce.create (key, value)` or `MessageToProduce.createWithHeaders headers (key, value)`, then `Producer.produce` (batch) or `Producer.produceSingle` (produce + flush). See `examples.md` → Produce With Key And Headers.
+- Choose `MessageKey.Simple` for a single-field key and `MessageKey.Delimited` when the key is a composite that must stay KSQL-compatible.
+- Lag monitoring: `Admin.lags` returns `PartitionLag list`; sum with `List.sumBy PartitionLag.lag`. See `examples.md` → Consumer Lag.
+
+## Error Handling
+
+- The API is railway-oriented: consume/commit results are `Result` / `AsyncResult` carrying DU error types (`ConsumeError`, `ManualCommitError`, `MetaDataParseError`). Match on the specific cases rather than catching exceptions.
+- `ConsumeError.PreviousMessageWasNotCommited` only appears under manual commit with `FailOnNotCommittedMessage.WithException`; it signals the previous message was never committed.
+- Under manual commit, always inspect the `Result` returned by `ManualCommit.execute` and surface `ManualCommitError` instead of ignoring it.
+
+## Composition
+
+- Transform payloads while preserving trace and commit handles with `TracedMessage.map`. This keeps the `Commit` handle attached after parsing.
+- When you start child spans for processing, finish them — `TracedMessage.finish` finishes the message's own span; spans you start from `Alma.Tracing` you finish yourself.
+
+## Integration with Other Libraries
+
+- Tracing is automatic only when a tracer is active in the host process; the library checks tracer availability and otherwise produces an inactive span. The trace context is propagated as B3 Kafka headers — the consumer extracts it, the producer injects it. No manual header plumbing is required for propagation.
+- External checkpoints integrate via `GetCheckpoint = Some f` on the consumer configuration, where `f: GroupId -> TopicPartition -> AsyncResult<TopicPartitionOffset, exn>`. See `examples.md` → External Checkpoint.
+- Health checks come from `Checker.defaultChecker` / `IntervalChecker.defaultChecker`; attach them to the configuration's `Checker` / `IntervalChecker` fields to gate consuming/producing on cluster and topic availability with retry and `Alma.Metrics` status marking.
+
+## Naming Conventions
+
+- Single-case DU wrappers (`BrokerList of string`, `Offset of int64`, `EventId of Guid`, …) each have a `[<RequireQualifiedAccess>]` companion module exposing `value` and constructors. Follow this module-per-type convention for any helper you add.
+- Namespace is `Alma.Kafka`; qualify modules (`Consumer.`, `Producer.`, `Admin.`) rather than opening everything.
+
+## Testing Recommendations
+
+- Use `GroupId.Random` in tests to force reading a topic from the beginning with an isolated group.
+- The repository's own automated coverage is limited to trace-propagation tests; consumer/producer behavior against a real broker is verified manually (the `example/` project ships a `docker-compose.yaml` with Kafka for local runs). Write integration tests against a disposable broker when changing consume/produce logic.

.claude/skills

@@ -0,0 +1 @@
+../.agents/skills

AGENTS.md

@@ -1,5 +1,7 @@
 # AGENTS.md — Alma.Kafka (fkafka)
 
+This repo ships Agent Skill for the `Alma.Kafka` library. Compatible agents discover it automatically; see `.agents/skills/fkafka/SKILL.md`.
+
 ## Project Purpose
 
 F# library (`Alma.Kafka`) for producing and consuming messages to/from Apache Kafka streams. Provides a typed, traced API with consumer lag monitoring, health checking, manual/auto commit modes, external checkpoint support, event schema definitions, metadata parsing, and trace propagation via Kafka headers. Published as a NuGet package.

CLAUDE.md

@@ -0,0 +1,2 @@
+<!-- Claude Code does not read AGENTS.md natively; this import bridges it. -->
+@AGENTS.md