Possible Architecture inspired by Flower

# Async Federated Learning with Tekton — Architecture

## Overview

This document describes an async federated learning system built on Tekton and an OCI-compatible
model registry. It borrows the client and strategy interface style from Flower but does not modify
or depend on the Flower codebase. Orchestration is fully event-driven and coordinator-free: OCI
registry webhooks trigger Tekton PipelineRuns, each PipelineRun reads a per-participant metadata
store to decide whether work should proceed, and all state transitions are driven by OCI artifact
pushes. There is no always-on coordinator process.

Each collaborator runs in its own independent Kubernetes cluster. The OCI registry is the sole
cross-cluster communication channel — no k8s API access crosses cluster boundaries. Each cluster
manages its own local metadata (Kubernetes ConfigMaps) and its own Tekton EventListener scoped to
events relevant to that participant.

The design is informed by the PAPAYA async FL system (arxiv.org/abs/2111.04877): no round
barriers, staleness-discounted aggregation, hybrid time/quantity triggering, max staleness
cutoff, and differential privacy at the aggregator.

---

## Background: [Flower](https://github.com/flwrlabs/flower)'s Federated Learning Model

The interface contracts in this system are modelled directly on Flower's design. Understanding the
original is useful context.

### Flower Core Abstractions

**`Client`** — what each collaborator implements:

| Method | Purpose |
|---|---|
| `get_parameters()` | Return current local model weights |
| `fit(ins)` | Train on local data, return updated weights + num_examples + metrics |
| `evaluate(ins)` | Evaluate model on local data, return loss + num_examples + metrics |

**`Strategy`** — what the aggregator implements:

| Method | Purpose |
|---|---|
| `initialize_parameters()` | Provide initial global model |
| `configure_fit()` | Select clients and send training instructions |
| `aggregate_fit()` | Merge client updates into new global model |
| `configure_evaluate()` | Select clients for evaluation |
| `aggregate_evaluate()` | Aggregate evaluation results |
| `evaluate()` | Optional centralized server-side evaluation |

**Flower's synchronous round barrier** (what we are replacing):

```
for each round:
  1. configure_fit()         → select clients, send global params
  2. fit_clients()           → ThreadPoolExecutor, wait for ALL to respond
  3. aggregate_fit()         → merge updates
  4. evaluate_round()        → optional distributed evaluation
```

All existing Flower strategies are synchronous — every round waits for a quorum before
aggregating. The `FaultTolerantFedAvg` and `round_timeout` give straggler tolerance but
the round barrier remains.

### Why Not Use Flower's Grid API

Flower's `Grid` API (`push_messages` / `pull_messages`) does support async patterns via
polling and message TTLs, but it requires the ServerApp and SuperNode processes to be
always-on and communicates via gRPC through the SuperLink. For a Kubernetes-native,
Tekton-based system we want ephemeral compute (PipelineRuns), not long-running processes,
and object-store-driven coordination rather than gRPC message routing.

---

## System Design Goals

- **No modifications to Flower** — borrow the interface contract, not the code
- **No API tie-in** — OCI Distribution Spec (not S3) as the storage API standard
- **True async, no rounds** — each collaborator runs independently; the model version
  counter is the only clock (PAPAYA-style)
- **No coordinator process** — state lives in cluster-local ConfigMaps and OCI; decisions
  are made inside PipelineRuns via `when` expressions
- **Multi-cluster isolation** — each collaborator runs in its own k8s cluster; the OCI
  registry is the only shared infrastructure; no cross-cluster k8s API access
- **Ephemeral compute** — Tekton PipelineRuns, not always-on processes
- **Kubernetes-native** — GPU scheduling, resource limits, RBAC via standard k8s primitives
- **Idempotent by design** — every PipelineRun checks metadata before acting; duplicate
  triggers are harmless no-ops

---

## No Rounds — PAPAYA's Core Principle

In synchronous FL, a "round" is a hard barrier: all clients train on model `vN`, submit,
server aggregates, everyone moves to `vN+1` together. Round number and model version are
the same concept.

In this system (following PAPAYA), the global model version is a **monotonic counter that
increments every time aggregation fires**. At any moment:

- `hospital-a` might be training on `v7`
- `hospital-b` might still be training on `v4` (slow, started earlier)
- `hospital-c` might have submitted `v7` and already started on `v8`

These are not "different rounds" — they are clients operating at their own pace. When
aggregation fires using `{hospital-a, hospital-c}` updates:

- `hospital-a` and `hospital-c` are idle — give them new work on `v8` immediately
- `hospital-b` is in-flight on `v4` — leave it alone; its stale update will be
  staleness-discounted when it eventually submits

The per-collaborator metadata store makes this explicit and prevents duplicate launches.

---

## Multi-Cluster Topology

```
┌─────────────────────────────────────────────────────────────────────────────┐
│  OCI Registry  (Quay / Harbor / ghcr.io)                                    │
│  — the only shared infrastructure across all clusters —                     │
│                                                                             │
│  fl-global              vN              global model weights                │
│  fl-updates             {client}-vN     collaborator model updates          │
│  fl-client-metrics      {client}-vN     per-run training metrics (JSON)     │
│  fl-aggregation-triggers vN             threshold-met marker                │
│  fl-train-signals       vN-{client}     per-collaborator re-launch signal   │
│  fl-time-checks         latest          CronJob heartbeat for time trigger  │
└──────────────────────────────────────────────────────────────────────────────┘
         ▲  ▼ push/pull (oras)               ▲  ▼ push/pull (oras)
         │  webhooks to each cluster          │  webhooks to aggregator cluster
         │                                    │
┌────────┴──────────────────┐     ┌──────────┴───────────────────────────────┐
│  Hospital-A Cluster       │     │  Aggregator Cluster                       │
│                           │     │                                           │
│  EventListener            │     │  EventListener                            │
│    Trigger C:             │     │    Trigger A: fl-updates push             │
│    fl-train-signals       │     │      → AggregationCheckPipelineRun        │
│    *-hospital-a           │     │    Trigger B: fl-aggregation-triggers     │
│    → TrainingPipelineRun  │     │      → AggregationPipelineRun             │
│                           │     │    Trigger D: fl-time-checks              │
│  ConfigMap:               │     │      → AggregationCheckPipelineRun        │
│    fl-meta-hospital-a     │     │                                           │
│    (cluster-local)        │     │  ConfigMap:                               │
│                           │     │    fl-meta-aggregator                     │
│  TrainingPipelineRun      │     │    (cluster-local)                        │
│    reads/writes own       │     │                                           │
│    ConfigMap only         │     │  AggregationCheckPipelineRun              │
│                           │     │  AggregationPipelineRun                   │
│  GPU nodes, local data    │     │    reads/writes own ConfigMap only        │
└───────────────────────────┘     └───────────────────────────────────────────┘

┌───────────────────────────┐
│  Hospital-B Cluster       │  (identical structure to Hospital-A)
│  Hospital-C Cluster       │
│  ...                      │
└───────────────────────────┘
```

**The critical constraint:** no k8s API call crosses a cluster boundary. The aggregator
never patches a collaborator's ConfigMap. A collaborator never creates a PipelineRun in
another cluster. All cross-cluster communication is OCI push events.

---

## Metadata Store (Per Participant, Cluster-Local)

Mutable coordination state lives in a **Kubernetes ConfigMap per participant**, local to
that participant's cluster. OCI artifacts are immutable content-addressed storage with no
atomic read-modify-write semantics. ConfigMaps are k8s-native, and their `resourceVersion`
field provides optimistic concurrency — two simultaneous writers, one gets a 409 Conflict
and retries.

Because ConfigMaps are cluster-local, each cluster manages its own state independently.
The aggregator never needs to read or write a collaborator's ConfigMap, and vice versa.

### Per-collaborator ConfigMap (lives in collaborator cluster)

```yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: fl-meta-hospital-a
  namespace: fl
data:
  in_flight:              "false"
  last_trained_version:   "7"
  last_submitted_version: "7"
```

### Aggregator ConfigMap (lives in aggregator cluster)

```yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: fl-meta-aggregator
  namespace: fl
data:
  current_global_version:   "8"
  aggregation_in_progress:  "false"
  aggregation_threshold:    "3"
  max_staleness:            "10"
```

---

## OCI Registry as Model Store and Cross-Cluster Bus

### Why OCI over S3/MinIO

| Property | OCI Distribution Spec | S3 API |
|---|---|---|
| Standardization | IANA/ISO open standard | Amazon's API (de-facto standard) |
| Vendor portability | Quay, Harbor, ghcr.io, ECR, GCR — identical client code | MinIO, S3, GCS compat — mostly compatible |
| Versioning | Tags + SHA256 content digests, first-class | Object keys + ETags, manual |
| Integrity verification | Automatic digest verification on push/pull | Manual checksumming |
| Metadata | Manifest annotations — no sidecar files | Object metadata or sidecar files |
| Notifications | Repository push webhooks | Object-level PUT events |
| Multi-cluster | Single registry serves all clusters; webhooks fan out to each | Same, but S3 event routing is less flexible |

### Repository Structure

```
quay.io/myorg/
├── fl-global              ← aggregator writes; tags: v0, v1, v2 ...
├── fl-updates             ← all collaborators write; tags: {client-id}-v{N}
│                            e.g. hospital-a-v7, hospital-b-v4, hospital-c-v7
├── fl-client-metrics      ← collaborators write per-run metrics; tags: {client-id}-v{N}
│                            small JSON artifact, one per training run
├── fl-aggregation-triggers← aggregator writes when threshold met; tags: v{N}
├── fl-train-signals       ← aggregator writes after aggregation; tags: v{N}-{client-id}
└── fl-time-checks         ← CronJob writes for hybrid time trigger; tags: latest
```

### Manifest Annotations

Training metadata lives on the `fl-updates` OCI manifest — no sidecar files:

```python
oras_client.push(
    files=["model_weights.npz"],
    target="quay.io/myorg/fl-updates:hospital-a-v7",
    manifest_annotations={
        "fl.base-version":  "7",
        "fl.num-examples":  "4821",
        "fl.client-id":     "hospital-a",
        "fl.timestamp":     "2026-03-14T10:22:00Z",
    }
)
```

### Per-Run Metrics Artifact

Training metrics from `client.fit()` are pushed as a separate small JSON artifact to
`fl-client-metrics`. This is the cross-cluster-safe metrics store: the aggregator can
pull it during aggregation without any network path beyond the OCI registry.

```python
oras_client.push(
    files=["metrics.json"],   # {"train_loss": 0.234, "train_accuracy": 0.891, ...}
    target="quay.io/myorg/fl-client-metrics:hospital-a-v7",
    artifact_type="application/vnd.fl.client-metrics",
    manifest_annotations={
        "fl.client-id":    "hospital-a",
        "fl.base-version": "7",
        "fl.num-examples": "4821",
        "fl.timestamp":    "2026-03-14T10:22:00Z",
    }
)
```

The run identity is `(client_id, base_version)` — the same tag used for the model update.
No separate run ID system required. The aggregation pipeline can optionally pull these
metrics artifacts alongside model updates to compute weighted aggregate metrics
(e.g. weighted average training loss across contributors).

### OCI Layering in the MVP

Full models are pushed as a **single blob layer** per artifact. OCI layer deduplication
only helps when layers are byte-for-byte identical across pushes. Since gradient descent
updates all weights each round, no round-over-round savings apply to fully trainable models.

OCI layering becomes meaningful post-MVP:
- **Frozen backbone**: frozen layer hash never changes, never re-uploaded after first push
- **Delta compression**: sparse per-block layers deduplicate unchanged blocks

For the MVP: single blob, no layer decomposition.

---

## High-Level Event Flow

```
┌──────────────────────────────────────────────────────────────────────────────┐
│  OCI Registry — sends webhooks to registered endpoint per repository         │
└──┬──────────────────────────────────────┬──────────────────────────────────┘
   │ fl-updates push                       │ fl-train-signals push
   │ fl-aggregation-triggers push          │ (one per contributing client)
   │ fl-time-checks push                   │
   ▼                                       ▼
Aggregator Cluster EventListener      Hospital-A Cluster EventListener
  Trigger A → AggCheckPipelineRun       Trigger C (CEL: *-hospital-a only)
  Trigger B → AggPipelineRun              → TrainingPipelineRun
  Trigger D → AggCheckPipelineRun
                │                               │
                │ (all within aggregator         │ (all within hospital-a
                │  cluster only)                 │  cluster only)
                ▼                               ▼
         AggCheckPipelineRun           TrainingPipelineRun
           reads fl-meta-aggregator      reads fl-meta-hospital-a
           counts fl-updates tags        when(should_train):
           when(threshold_met):            ClaimTraining
             ClaimAggregation              Train
             PushAggTrigger →OCI           pushes fl-updates:hospital-a-vN
                │                         pushes fl-client-metrics:hospital-a-vN
                ▼                         ReleaseMetadata (finally)
         AggPipelineRun
           Aggregate
           pushes fl-global:vN+1
           ReleaseAndSignal:
             patches fl-meta-aggregator (local)
             pushes fl-train-signals:vN+1-hospital-a → OCI
             pushes fl-train-signals:vN+1-hospital-c → OCI
             (hospital-b not in batch, left alone)
```

---

## Pipeline Definitions

### TrainingPipeline (runs in collaborator cluster)

Triggered by `fl-train-signals:vN-{client-id}` push (Trigger C, collaborator EventListener):

```
Task: ReadMetadata
  kubectl get configmap fl-meta-{client-id}   ← cluster-local, always accessible
  outputs:
    in_flight            → "false"
    last_trained_version → "7"
    should_train         → "true"  (new_version > last_trained AND not in_flight)

when (should_train == "true"):

  Task: ClaimTraining
    kubectl patch configmap fl-meta-{client-id}
      in_flight=true, last_trained_version={new_version}
    resourceVersion ensures only one claimer wins; 409 = another PipelineRun
    already claimed, skip remaining tasks

  Task: Train  (runAfter: ClaimTraining)
    oras pull fl-global:v{new_version}         ← from shared OCI registry
    client.fit(params, config)
    oras push fl-updates:{client-id}-v{new_version}
      manifest_annotations(num_examples, timestamp)
    oras push fl-client-metrics:{client-id}-v{new_version}
      metrics.json (train_loss, train_accuracy, etc. from fit() return)
    ← both OCI pushes fire webhooks to aggregator's EventListener (Trigger A)

Task: ReleaseMetadata  (finally — always runs regardless of Train outcome)
  kubectl patch configmap fl-meta-{client-id}
    in_flight=false, last_submitted_version={new_version}
  ← cluster-local, no cross-cluster access needed
```

**Self-contained:** the collaborator cluster reads and writes only its own ConfigMap.
The aggregator is never involved in collaborator metadata management.

If `in_flight=true` when `ReadMetadata` runs, `should_train=false` and the `when` guard
skips `ClaimTraining` and `Train`. Duplicate trigger = harmless no-op.

### AggregationCheckPipeline (runs in aggregator cluster)

Triggered by `fl-updates:{client}-vN` push (Trigger A) or `fl-time-checks` push (Trigger D):

```
Task: ReadAggregatorMeta
  kubectl get configmap fl-meta-aggregator   ← aggregator cluster-local
  outputs:
    current_version          → "8"
    aggregation_in_progress  → "false"
    threshold                → "3"
    max_staleness            → "10"

Task: CountUpdates  (runAfter: ReadAggregatorMeta)
  oras tag list quay.io/myorg/fl-updates \
    | grep -E "^[a-z-]+-v{current_version}$" | wc -l
  outputs:
    update_count   → "3"
    threshold_met  → "true"

  # PAPAYA hybrid trigger
  if time_triggered=true AND update_count > 0:
    threshold_met = "true"

  # PAPAYA max staleness: exclude updates beyond threshold
  eligible_updates → '[{"client_id":"hospital-a","base_version":"8","staleness":0},...]'

when (threshold_met == "true" AND aggregation_in_progress == "false"):

  Task: ClaimAggregation
    kubectl patch configmap fl-meta-aggregator
      aggregation_in_progress=true
    resourceVersion — 409 on conflict means another PipelineRun already claimed; exit

  Task: PushAggregationTrigger  (runAfter: ClaimAggregation)
    oras push quay.io/myorg/fl-aggregation-triggers:v{current_version}
      manifest_annotations(eligible_updates=...)
    ← fires webhook to aggregator EventListener (Trigger B)
```

### AggregationPipeline (runs in aggregator cluster)

Triggered by `fl-aggregation-triggers:vN` push (Trigger B, aggregator EventListener):

```
Task: Aggregate
  # Idempotency: exit if fl-global:v{N+1} already exists
  if oras tag list fl-global | grep "^v{N+1}$"; then exit 0; fi

  # Pull eligible updates and their metrics from shared OCI registry
  for each update in eligible_updates:
    oras pull fl-updates:{client_id}-v{base_version}      ← model weights
    oras pull fl-client-metrics:{client_id}-v{base_version} ← training metrics
    staleness = current_version - base_version

  # Pull base global model
  oras pull fl-global:v{current_version}

  # Staleness-discounted aggregation
  strategy.aggregate_fit(global_params, [(weights, n_examples, metrics, staleness), ...])

  # PAPAYA Differential Privacy
  noise = np.random.normal(0, dp_noise_scale, shape)
  new_params = aggregated_params + noise

  # Push new global model
  oras push fl-global:v{N+1}

Task: ReleaseAndSignal  (runAfter: Aggregate)
  # Update aggregator's own local metadata only
  kubectl patch configmap fl-meta-aggregator
    aggregation_in_progress=false, current_global_version={N+1}
  ← cluster-local, no cross-cluster k8s access

  # Signal contributing clients via OCI — cross-cluster safe
  for each contributing client_id:
    oras push fl-train-signals:v{N+1}-{client_id}
    ← fires webhook to that client's cluster EventListener (Trigger C)

  # Non-contributing (in-flight) clients are not signalled
  # Their own ReleaseMetadata (finally) task manages their local ConfigMap
  # They will submit their stale update to the next aggregation
```

**Key property:** `ReleaseAndSignal` only touches the aggregator's own ConfigMap and
pushes to OCI. It never issues a k8s API call to any collaborator cluster.

---

## Event Flow (One Complete Async Cycle)

```
Bootstrap: in each collaborator cluster, kubectl create pipelinerun
           pointing at fl-train-signals:v0-{client-id}

① TrainingPipeline — hospital-a cluster, model_version=8
    ReadMetadata   → in_flight=false, should_train=true  (cluster-local ConfigMap)
    ClaimTraining  → patches fl-meta-hospital-a: in_flight=true  (cluster-local)
    Train          → pulls fl-global:v8 from OCI
                   → client.fit()
                   → pushes fl-updates:hospital-a-v8 to OCI
                   → pushes fl-client-metrics:hospital-a-v8 to OCI
    ReleaseMetadata (finally) → patches fl-meta-hospital-a: in_flight=false (cluster-local)
         │
         │ OCI webhook on fl-updates fires to aggregator cluster endpoint
         ▼
② AggregationCheckPipeline — aggregator cluster
    ReadAggregatorMeta → current_version=8, not_in_progress  (cluster-local ConfigMap)
    CountUpdates       → count=1, threshold=3, threshold_met=false
    when guard fails → exits (no-op)

   (hospital-b and hospital-c complete in their own clusters; OCI webhooks fire)

③ AggregationCheckPipeline — triggered by hospital-c-v8 push
    CountUpdates → count=3, threshold_met=true
    ClaimAggregation → patches fl-meta-aggregator: in_progress=true  (cluster-local)
    PushAggregationTrigger → pushes fl-aggregation-triggers:v8 to OCI
         │
         │ OCI webhook fires to aggregator cluster endpoint
         ▼
④ AggregationPipeline — aggregator cluster
    Aggregate  → checks fl-global:v9 not yet exists (idempotency)
               → pulls fl-updates and fl-client-metrics for each eligible update from OCI
               → staleness: hospital-a=0, hospital-b=2, hospital-c=0
               → strategy.aggregate_fit() with staleness discounting
               → adds DP noise
               → pushes fl-global:v9 to OCI

    ReleaseAndSignal
               → patches fl-meta-aggregator: in_progress=false, version=9  (cluster-local)
               → pushes fl-train-signals:v9-hospital-a to OCI
               → pushes fl-train-signals:v9-hospital-b to OCI
               → pushes fl-train-signals:v9-hospital-c to OCI
         │
         │ OCI webhooks fire to each collaborator cluster's EventListener endpoint
         ▼
⑤ Each collaborator cluster independently:
    hospital-a EventListener (CEL: *-hospital-a) → TrainingPipelineRun in hospital-a cluster
    hospital-b EventListener (CEL: *-hospital-b) → TrainingPipelineRun in hospital-b cluster
    hospital-c EventListener (CEL: *-hospital-c) → TrainingPipelineRun in hospital-c cluster

→ each runs independently → back to ①
```

---

## EventListener Configuration

Each cluster runs its own Tekton EventListener scoped to events relevant to that
participant. The OCI registry sends webhooks for a given repository to all registered
endpoints. CEL interceptors discard irrelevant events.

### Aggregator cluster EventListener

```yaml
apiVersion: triggers.tekton.dev/v1beta1
kind: EventListener
metadata:
  name: fl-listener
  namespace: fl
spec:
  serviceAccountName: fl-trigger-sa
  triggers:

    # Trigger A: collaborator pushed an update
    - name: update-written
      interceptors:
        - ref:
            name: cel
          params:
            - name: filter
              value: "body.repository == 'fl-updates'"
            - name: overlays
              value:
                - key: client_id
                  expression: "body.updated_tags[0].split('-v')[0]"
                - key: base_version
                  expression: "body.updated_tags[0].split('-v')[1]"
      bindings:
        - ref: trigger-a-binding
      template:
        ref: aggregation-check-pipeline-template

    # Trigger B: threshold marker pushed → run aggregation
    - name: aggregation-triggered
      interceptors:
        - ref:
            name: cel
          params:
            - name: filter
              value: "body.repository == 'fl-aggregation-triggers'"
            - name: overlays
              value:
                - key: current_version
                  expression: "body.updated_tags[0].split('v')[1]"
      bindings:
        - ref: trigger-b-binding
      template:
        ref: aggregation-pipeline-template

    # Trigger D: time-check → run aggregation check in time-triggered mode
    - name: time-check
      interceptors:
        - ref:
            name: cel
          params:
            - name: filter
              value: "body.repository == 'fl-time-checks'"
      bindings:
        - ref: trigger-d-binding
      template:
        ref: aggregation-check-pipeline-template
```

### Collaborator cluster EventListener (hospital-a shown)

```yaml
apiVersion: triggers.tekton.dev/v1beta1
kind: EventListener
metadata:
  name: fl-listener
  namespace: fl
spec:
  serviceAccountName: fl-trigger-sa
  triggers:

    # Trigger C: train signal for this collaborator only
    - name: train-signal-received
      interceptors:
        - ref:
            name: cel
          params:
            # CEL filter: only act on signals addressed to this collaborator
            - name: filter
              value: "body.repository == 'fl-train-signals'
                      && body.updated_tags[0].endsWith('-hospital-a')"
            - name: overlays
              value:
                - key: new_version
                  expression: "body.updated_tags[0].split('-')[0].split('v')[1]"
                - key: client_id
                  expression: "body.updated_tags[0].split('-', 1)[1]"
      bindings:
        - ref: trigger-c-binding
      template:
        ref: training-pipeline-template
```

The `endsWith('-hospital-a')` filter means this EventListener ignores train signals for
hospital-b, hospital-c, etc. even though the OCI registry sends the webhook to all
registered endpoints for the `fl-train-signals` repository.

---

## Staleness-Aware Aggregation

Since clients train asynchronously they may train on different global model versions.
A client that trained on `v6` while the global model is now at `v8` has staleness `2`.

**Delta-based aggregation is more principled than full-weight blending for stale updates:**

```
# Full-weight blending (problematic):
new_global = 0.7 * global_v8 + 0.3 * client_weights_trained_on_v6
# Anchors part of the model to an outdated point in time

# Delta aggregation (correct):
delta_A = client_weights_A - global_v6     # what the client learned
staleness = current_version - base_version
weight = num_examples / (1 + staleness)

new_global = global_v8 + Σ(weight_i * delta_i) / Σ(weight_i)
```

**MVP uses full model weights** (no delta compression). The aggregation pipeline computes
deltas internally by subtracting the base model from the client update.

**Max staleness cutoff (PAPAYA):** updates with `staleness > max_staleness` are excluded
from `eligible_updates` in `CountUpdates`. The client is re-signalled with the current
model version so it restarts training on a fresh base.

---

## PAPAYA Features Incorporated

| Feature | Where implemented | Status |
|---|---|---|
| Async aggregation, no round barrier | PipelineRun-per-client, independent clusters | MVP |
| Staleness-discounted aggregation `1/(1+s)` | `strategy.aggregate_fit()` | MVP |
| Quantity threshold trigger | `CountUpdates` task | MVP |
| Time window trigger (hybrid) | CronJob → `fl-time-checks` → Trigger D | MVP |
| Max staleness cutoff | `CountUpdates` excludes stale updates | MVP |
| Differential Privacy (Gaussian noise) | `Aggregate` task after `aggregate_fit()` | MVP |
| Per-run collaborator metrics | `fl-client-metrics` OCI artifact per training run | MVP |
| Secure Aggregation | Requires pre-training key agreement; conflicts with independent-cluster model | Post-MVP |
| Adaptive client prioritisation | Requires per-client history tracking in aggregator | Post-MVP |

---

## Interfaces (Borrowed from Flower, No Dependency)

### `interfaces/client.py`

```python
from abc import ABC, abstractmethod
import numpy as np

NDArrays = list[np.ndarray]
Scalar   = int | float | str | bool

class NumPyClient(ABC):
    """Implement this on each collaborator node."""

    @abstractmethod
    def get_parameters(self) -> NDArrays:
        """Return initial model weights."""

    @abstractmethod
    def fit(
        self,
        parameters: NDArrays,
        config: dict[str, Scalar],
    ) -> tuple[NDArrays, int, dict[str, Scalar]]:
        """Train on local data.

        Returns
        -------
        parameters  : updated model weights
        num_examples: number of training examples used
        metrics     : arbitrary scalar metrics (loss, accuracy, ...)
                      stored as fl-client-metrics OCI artifact
        """

    @abstractmethod
    def evaluate(
        self,
        parameters: NDArrays,
        config: dict[str, Scalar],
    ) -> tuple[float, int, dict[str, Scalar]]:
        """Evaluate model on local data.

        Returns
        -------
        loss        : scalar loss value
        num_examples: number of evaluation examples used
        metrics     : arbitrary scalar metrics
        """
```

### `interfaces/strategy.py`

```python
from abc import ABC, abstractmethod
from functools import reduce
import numpy as np

class Strategy(ABC):

    @abstractmethod
    def initialize_parameters(self) -> NDArrays:
        """Return the initial global model weights."""

    @abstractmethod
    def aggregate_fit(
        self,
        global_params: NDArrays,
        results: list[tuple[NDArrays, int, dict, int]],
        # each entry: (client_weights, num_examples, metrics, staleness)
    ) -> NDArrays:
        """Aggregate client updates into a new global model."""

    @abstractmethod
    def aggregate_evaluate(
        self,
        results: list[tuple[float, int, dict]],
        # each entry: (loss, num_examples, metrics)
    ) -> tuple[float, dict]:
        """Aggregate evaluation results."""


class FedAvg(Strategy):
    """Staleness-discounted federated averaging."""

    def aggregate_fit(self, global_params, results):
        total_weight = sum(n / (1 + s) for _, n, _, s in results)
        aggregated_delta = [
            reduce(
                np.add,
                [
                    (weights[i] - global_params[i]) * (num_examples / (1 + staleness))
                    for weights, num_examples, _, staleness in results
                ],
            ) / total_weight
            for i in range(len(global_params))
        ]
        return [g + d for g, d in zip(global_params, aggregated_delta)]

    def aggregate_evaluate(self, results):
        total = sum(n for _, n, _ in results)
        loss = sum(l * n for l, n, _ in results) / total
        return loss, {}
```

---

## Directory Structure

```
fl-tekton/
│
├── interfaces/
│   ├── client.py                   # NumPyClient ABC
│   └── strategy.py                 # Strategy ABC + FedAvg
│
├── runtime/
│   ├── train_runner.py             # Train task entrypoint
│   ├── aggregate_runner.py         # Aggregate task entrypoint
│   └── evaluate_runner.py          # Evaluate task entrypoint
│
├── store/
│   └── model_store.py              # ModelStore ABC + OCIModelStore
│
└── k8s/
    ├── aggregator/                 # deployed to aggregator cluster
    │   ├── pipelines/
    │   │   ├── aggregation-check-pipeline.yaml
    │   │   └── aggregation-pipeline.yaml
    │   ├── tasks/
    │   │   ├── fl-read-aggregator-meta-task.yaml
    │   │   ├── fl-count-updates-task.yaml
    │   │   ├── fl-claim-aggregation-task.yaml
    │   │   ├── fl-push-aggregation-trigger-task.yaml
    │   │   ├── fl-aggregate-task.yaml
    │   │   └── fl-release-and-signal-task.yaml
    │   ├── triggers/
    │   │   ├── event-listener.yaml           # Triggers A, B, D
    │   │   ├── trigger-a-binding.yaml
    │   │   ├── trigger-b-binding.yaml
    │   │   ├── trigger-d-binding.yaml
    │   │   ├── aggregation-check-template.yaml
    │   │   └── aggregation-pipeline-template.yaml
    │   ├── config/
    │   │   └── fl-meta-aggregator.yaml
    │   ├── rbac/
    │   │   ├── trigger-sa.yaml
    │   │   ├── pipeline-sa.yaml
    │   │   └── oci-registry-secret.yaml
    │   └── cron/
    │       └── fl-time-check-cronjob.yaml
    │
    └── collaborator/               # deployed to each collaborator cluster
        ├── pipelines/
        │   └── training-pipeline.yaml
        ├── tasks/
        │   ├── fl-read-metadata-task.yaml
        │   ├── fl-claim-training-task.yaml
        │   ├── fl-train-task.yaml
        │   └── fl-release-metadata-task.yaml
        ├── triggers/
        │   ├── event-listener.yaml           # Trigger C only, CEL-filtered per client
        │   ├── trigger-c-binding.yaml
        │   └── training-pipeline-template.yaml
        ├── config/
        │   └── fl-meta-collaborator.yaml     # parameterised; client_id set at deploy time
        └── rbac/
            ├── trigger-sa.yaml
            ├── pipeline-sa.yaml
            └── oci-registry-secret.yaml      # push: fl-updates, fl-client-metrics only
```

The `collaborator/` directory is deployed identically to each collaborator cluster with
one parameter substituted: the `client_id` value in the ConfigMap and the CEL filter in
the EventListener. Everything else is identical across collaborators.

---

## Component Detail

### `store/model_store.py`

```python
class ModelStore(ABC):

    @abstractmethod
    def push_global(self, version: int, arrays: NDArrays) -> str:
        """Push full global model. Returns OCI digest."""

    @abstractmethod
    def pull_global(self, version: int) -> NDArrays:
        """Pull global model by version tag."""

    @abstractmethod
    def push_update(
        self,
        client_id: str,
        base_version: int,
        arrays: NDArrays,
        num_examples: int,
    ) -> str:
        """Push full client model. Returns OCI digest."""

    @abstractmethod
    def push_metrics(
        self,
        client_id: str,
        base_version: int,
        metrics: dict,
        num_examples: int,
    ) -> str:
        """Push per-run training metrics as OCI artifact. Returns digest."""

    @abstractmethod
    def pull_update(
        self,
        client_id: str,
        base_version: int,
    ) -> tuple[NDArrays, int]:
        """Pull client update. Returns (arrays, num_examples)."""

    @abstractmethod
    def pull_metrics(
        self,
        client_id: str,
        base_version: int,
    ) -> dict:
        """Pull per-run training metrics for a client update."""

    @abstractmethod
    def pull_updates_batch(
        self,
        refs: list[dict],
    ) -> list[tuple[NDArrays, int, dict, int]]:
        """Pull updates + metrics. Returns (arrays, num_examples, metrics, staleness)."""

    @abstractmethod
    def global_version_exists(self, version: int) -> bool:
        """Check if fl-global:vN tag exists (idempotency check)."""

    @abstractmethod
    def list_updates_for_version(self, version: int) -> list[str]:
        """List client IDs that have submitted updates for a given version."""
```

### `runtime/train_runner.py`

```
Args: --client-module  path.to.module:ClassName
      --model-version  N
      --client-id      hospital-a
      --config         '{"epochs": 5, "lr": 0.01}'

Steps:
  1. Build ModelStore from env (OCI_REGISTRY, OCI_NAMESPACE, credentials)
  2. Import and instantiate user's NumPyClient subclass
  3. Pull global model: store.pull_global(model_version)
  4. Call client.fit(params, config) → (updated_params, num_examples, metrics)
  5. Push model update: store.push_update(client_id, model_version, updated_params, num_examples)
  6. Push run metrics: store.push_metrics(client_id, model_version, metrics, num_examples)
  7. Exit  ← both OCI pushes fire webhooks to aggregator cluster EventListener (Trigger A)
```

### `runtime/aggregate_runner.py`

```
Args: --strategy-module   path.to.module:ClassName
      --current-version   N
      --eligible-updates  '[{"client_id":"hospital-a","base_version":"8","staleness":"0"},...]'
      --dp-noise-scale    0.01

Steps:
  1. Build ModelStore from env
  2. Idempotency check: exit if store.global_version_exists(current_version + 1)
  3. Import and instantiate user's Strategy subclass
  4. Pull current global: store.pull_global(current_version)
  5. Pull updates + metrics batch: store.pull_updates_batch(refs)
     → (arrays, num_examples, metrics, staleness) per entry
  6. Call strategy.aggregate_fit(global_params, results)
  7. Add Gaussian DP noise: new_params[i] += np.random.normal(0, dp_noise_scale, shape)
  8. Push new global: store.push_global(current_version + 1, new_params)
  9. Exit
```

---

## Tekton Trigger Paths Summary

| OCI push | Fires webhook to | Trigger | PipelineRun (cluster) | Decision |
|---|---|---|---|---|
| `fl-updates:{client}-vN` | aggregator cluster | A | AggregationCheckPipeline (aggregator) | Count updates; if threshold met, push aggregation trigger |
| `fl-client-metrics:{client}-vN` | aggregator cluster | A (same repo? or separate) | — | Metrics available for aggregator to pull; no separate trigger needed |
| `fl-aggregation-triggers:vN` | aggregator cluster | B | AggregationPipeline (aggregator) | Aggregate; push new global; push per-client train signals |
| `fl-train-signals:vN-{client}` | all collaborator clusters | C (CEL-filtered per cluster) | TrainingPipeline (collaborator cluster) | Read local metadata; if not in-flight, claim and train |
| `fl-time-checks:latest` | aggregator cluster | D | AggregationCheckPipeline (aggregator) | Same as Trigger A with `time_triggered=true` |
| Bootstrap | n/a | Manual `kubectl create pipelinerun` per collaborator cluster | TrainingPipeline (each collaborator cluster) | Starts initial training |

---

## Access Control

### OCI credentials (per cluster)

```
Aggregator cluster:
  push: fl-global, fl-aggregation-triggers, fl-train-signals
  pull: fl-updates (all client tags), fl-client-metrics (all client tags), fl-global

Hospital-A cluster:
  push: fl-updates (hospital-a-* tags only), fl-client-metrics (hospital-a-* tags only)
  pull: fl-global
  no access to: fl-aggregation-triggers, fl-train-signals, other clients' tags

Hospital-B, Hospital-C: (same pattern as Hospital-A, scoped to their own tag prefix)
```

### k8s RBAC (per cluster, cluster-local only)

```
Aggregator cluster:
  EventListener SA: create PipelineRuns in fl namespace
  Pipeline SA: read/patch fl-meta-aggregator ConfigMap
               no credentials to any collaborator cluster

Hospital-A cluster:
  EventListener SA: create PipelineRuns in fl namespace
  Pipeline SA: read/patch fl-meta-hospital-a ConfigMap only
               no credentials to aggregator cluster or other collaborator clusters

CronJob (aggregator cluster):
  push fl-time-checks:latest only
```

No cluster has k8s API credentials for any other cluster. The OCI registry is the
complete and sole cross-cluster interface.

---

## Network Requirements

```
OCI registry:
  - Reachable from all clusters (read/write)
  - Webhook endpoints reachable from OCI registry:
      aggregator cluster:   https://aggregator-el.example.com  (EventListener ingress)
      hospital-a cluster:   https://hospital-a-el.example.com  (EventListener ingress)
      hospital-b cluster:   https://hospital-b-el.example.com
      ...

Clusters do NOT need network connectivity to each other.
All inter-cluster communication is via the OCI registry.
```

Each cluster's EventListener Service requires an ingress endpoint reachable from the OCI
registry's webhook delivery. In an enterprise/air-gapped setup this is typically handled
via VPN, private link, or a shared internal network that all clusters and the registry
share. Clusters themselves need not peer with each other.

---

## Build Order

1. **`store/model_store.py`** — `ModelStore` ABC and `OCIModelStore` with `push_metrics`
   and `pull_metrics`. All runtime code depends on this.

2. **`interfaces/`** — `NumPyClient` and `Strategy` ABCs. Write toy implementations for
   local testing.

3. **`runtime/train_runner.py`** + **`runtime/aggregate_runner.py`** — wire interfaces
   to store. Test standalone.

4. **`k8s/aggregator/config/fl-meta-aggregator.yaml`** and
   **`k8s/collaborator/config/fl-meta-collaborator.yaml`** — deploy ConfigMaps to
   respective clusters. Verify `kubectl patch` with `resourceVersion`.

5. **`k8s/aggregator/tasks/`** and **`k8s/collaborator/tasks/`** — deploy tasks to each
   cluster. Test each with a standalone `TaskRun` before assembling into Pipelines.

6. **`k8s/collaborator/pipelines/training-pipeline.yaml`** — deploy to a test collaborator
   cluster. Trigger manually. Verify `when` skip on `in_flight=true`.

7. **OCI webhook config** — configure push notifications for all repositories to point at
   each cluster's EventListener ingress endpoint. Verify with `oras push` + webhook logs.

8. **`k8s/aggregator/triggers/event-listener.yaml`** and
   **`k8s/collaborator/triggers/event-listener.yaml`** — deploy to respective clusters.
   Verify CEL filtering with mock payloads.

9. **`k8s/aggregator/pipelines/`** — deploy aggregation pipelines. Verify idempotency:
   trigger twice, confirm second is a no-op via `global_version_exists` check.

10. **`k8s/aggregator/cron/fl-time-check-cronjob.yaml`** — deploy CronJob to aggregator
    cluster. Verify Trigger D fires.

11. **End-to-end test** — two toy collaborator clusters (random `fit()`), one aggregator
    cluster, threshold=2, shared OCI registry, watch the full cross-cluster loop run
    autonomously for 10+ cycles.

---

## Post-MVP Roadmap

| Feature | What changes |
|---|---|
| Delta compression (TopK sparsification) | `train_runner.py` computes delta before push; `aggregate_runner.py` applies deltas additively; error buffer stored per client in `fl-client-metrics` OCI artifact |
| OCI layer decomposition | `OCIModelStore.push_update()` splits model into per-block layers; frozen layers deduplicated across clients and rounds |
| Evaluation pipeline | `EvaluationPipeline` in collaborator cluster; `fl-eval-signals` OCI repo mirrors `fl-train-signals`; aggregator pulls eval metrics artifacts |
| Global model evaluation (PAPAYA-style) | Aggregator holds validation set; `Aggregate` task evaluates `fl-global:vN+1` centrally and stores result in `fl-agg-metrics` OCI artifact |
| Adaptive client prioritisation | `CountUpdates` reads per-client history from `fl-client-metrics`; biases which updates are included |
| Secure Aggregation | Requires cohort formation phase before training; aggregator coordinates key exchange via new OCI signal type; breaks pure independence of collaborator clusters |
| Dynamic client registration | New collaborator cluster deploys `k8s/collaborator/` with their `client_id`; OCI tag-prefix policy added; no aggregator changes |
| Multi-federation isolation | Separate OCI namespace + aggregator cluster per federation; collaborator clusters can participate in multiple federations with separate credentials |
| OCI Referrers API | Replace tag-listing in `CountUpdates` with `oras discover`; requires same-repository setup for referrers to work |


Method	Purpose
`get_parameters()`	Return current local model weights
`fit(ins)`	Train on local data, return updated weights + num_examples + metrics
`evaluate(ins)`	Evaluate model on local data, return loss + num_examples + metrics

Method	Purpose
`initialize_parameters()`	Provide initial global model
`configure_fit()`	Select clients and send training instructions
`aggregate_fit()`	Merge client updates into new global model
`configure_evaluate()`	Select clients for evaluation
`aggregate_evaluate()`	Aggregate evaluation results
`evaluate()`	Optional centralized server-side evaluation

Property	OCI Distribution Spec	S3 API
Standardization	IANA/ISO open standard	Amazon's API (de-facto standard)
Vendor portability	Quay, Harbor, ghcr.io, ECR, GCR — identical client code	MinIO, S3, GCS compat — mostly compatible
Versioning	Tags + SHA256 content digests, first-class	Object keys + ETags, manual
Integrity verification	Automatic digest verification on push/pull	Manual checksumming
Metadata	Manifest annotations — no sidecar files	Object metadata or sidecar files
Notifications	Repository push webhooks	Object-level PUT events
Multi-cluster	Single registry serves all clusters; webhooks fan out to each	Same, but S3 event routing is less flexible

Feature	Where implemented	Status
Async aggregation, no round barrier	PipelineRun-per-client, independent clusters	MVP
Staleness-discounted aggregation `1/(1+s)`	`strategy.aggregate_fit()`	MVP
Quantity threshold trigger	`CountUpdates` task	MVP
Time window trigger (hybrid)	CronJob → `fl-time-checks` → Trigger D	MVP
Max staleness cutoff	`CountUpdates` excludes stale updates	MVP
Differential Privacy (Gaussian noise)	`Aggregate` task after `aggregate_fit()`	MVP
Per-run collaborator metrics	`fl-client-metrics` OCI artifact per training run	MVP
Secure Aggregation	Requires pre-training key agreement; conflicts with independent-cluster model	Post-MVP
Adaptive client prioritisation	Requires per-client history tracking in aggregator	Post-MVP

OCI push	Fires webhook to	Trigger	PipelineRun (cluster)	Decision
`fl-updates:{client}-vN`	aggregator cluster	A	AggregationCheckPipeline (aggregator)	Count updates; if threshold met, push aggregation trigger
`fl-client-metrics:{client}-vN`	aggregator cluster	A (same repo? or separate)	—	Metrics available for aggregator to pull; no separate trigger needed
`fl-aggregation-triggers:vN`	aggregator cluster	B	AggregationPipeline (aggregator)	Aggregate; push new global; push per-client train signals
`fl-train-signals:vN-{client}`	all collaborator clusters	C (CEL-filtered per cluster)	TrainingPipeline (collaborator cluster)	Read local metadata; if not in-flight, claim and train
`fl-time-checks:latest`	aggregator cluster	D	AggregationCheckPipeline (aggregator)	Same as Trigger A with `time_triggered=true`
Bootstrap	n/a	Manual `kubectl create pipelinerun` per collaborator cluster	TrainingPipeline (each collaborator cluster)	Starts initial training

Feature	What changes
Delta compression (TopK sparsification)	`train_runner.py` computes delta before push; `aggregate_runner.py` applies deltas additively; error buffer stored per client in `fl-client-metrics` OCI artifact
OCI layer decomposition	`OCIModelStore.push_update()` splits model into per-block layers; frozen layers deduplicated across clients and rounds
Evaluation pipeline	`EvaluationPipeline` in collaborator cluster; `fl-eval-signals` OCI repo mirrors `fl-train-signals`; aggregator pulls eval metrics artifacts
Global model evaluation (PAPAYA-style)	Aggregator holds validation set; `Aggregate` task evaluates `fl-global:vN+1` centrally and stores result in `fl-agg-metrics` OCI artifact
Adaptive client prioritisation	`CountUpdates` reads per-client history from `fl-client-metrics`; biases which updates are included
Secure Aggregation	Requires cohort formation phase before training; aggregator coordinates key exchange via new OCI signal type; breaks pure independence of collaborator clusters
Dynamic client registration	New collaborator cluster deploys `k8s/collaborator/` with their `client_id`; OCI tag-prefix policy added; no aggregator changes
Multi-federation isolation	Separate OCI namespace + aggregator cluster per federation; collaborator clusters can participate in multiple federations with separate credentials
OCI Referrers API	Replace tag-listing in `CountUpdates` with `oras discover`; requires same-repository setup for referrers to work

Uh oh!

Possible Architecture inspired by Flower #1

Description

Async Federated Learning with Tekton — Architecture

Overview

Background: Flower's Federated Learning Model

Flower Core Abstractions

Why Not Use Flower's Grid API

System Design Goals

No Rounds — PAPAYA's Core Principle

Multi-Cluster Topology

Metadata Store (Per Participant, Cluster-Local)

Per-collaborator ConfigMap (lives in collaborator cluster)

Aggregator ConfigMap (lives in aggregator cluster)

OCI Registry as Model Store and Cross-Cluster Bus

Why OCI over S3/MinIO

Repository Structure

Manifest Annotations

Per-Run Metrics Artifact

OCI Layering in the MVP

High-Level Event Flow

Pipeline Definitions

TrainingPipeline (runs in collaborator cluster)

AggregationCheckPipeline (runs in aggregator cluster)

AggregationPipeline (runs in aggregator cluster)

Event Flow (One Complete Async Cycle)

EventListener Configuration

Aggregator cluster EventListener

Collaborator cluster EventListener (hospital-a shown)

Staleness-Aware Aggregation

PAPAYA Features Incorporated

Interfaces (Borrowed from Flower, No Dependency)

interfaces/client.py

interfaces/strategy.py

Directory Structure

Component Detail

store/model_store.py

runtime/train_runner.py

runtime/aggregate_runner.py

Tekton Trigger Paths Summary

Access Control

OCI credentials (per cluster)

k8s RBAC (per cluster, cluster-local only)

Network Requirements

Build Order

Post-MVP Roadmap

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

`interfaces/client.py`

`interfaces/strategy.py`

`store/model_store.py`

`runtime/train_runner.py`

`runtime/aggregate_runner.py`