You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by we...@apache.org on 2026/10/07 20:52:53 UTC
(flink-agents-demos) branch main updated: [feature] Add flink-stateful-ops-agent-demo: a stateful runbook agent for Flink 2.2 balanced tasks scheduling (#10)
This is an automated email from the ASF dual-hosted git repository.
weiqingy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents-demos.git
The following commit(s) were added to refs/heads/main by this push:
new 052bdd0 [feature] Add flink-stateful-ops-agent-demo: a stateful runbook agent for Flink 2.2 balanced tasks scheduling (#10)
052bdd0 is described below
commit 052bdd002b3a2b793633fbda752ccde570da3422
Author: Weiqing Yang <ya...@gmail.com>
AuthorDate: Wed Oct 7 13:52:46 2026 -0700
[feature] Add flink-stateful-ops-agent-demo: a stateful runbook agent for Flink 2.2 balanced tasks scheduling (#10)
A Flink Agents job that follows the stateful runbook in the Flink 2.2 Balanced
Tasks Scheduling docs (FLIP-370): it waits for skew that persists across
windows, applies TASKS through a real redeploy from a savepoint, checks the
result against a baseline kept in memory (KEEP or REVERT), raises
slot.request.max-interval step by step, and escalates to FLINK-38715 with the
attempt history. Each job's memory is keyed state, and side effects run as
durable actions with a reconciler, so killing the agent mid-redeploy does not
send the redeploy twice.
Includes the target cluster and ClickstreamEnrichment job, a platform simulator
with a dashboard, bin/demo.sh, and the recording runbook and demo script for
re:Invent 2026 OPN303.
Co-authored-by: Copilot <22...@users.noreply.github.com>
---
flink-stateful-ops-agent-demo/.gitignore | 11 +
.../agent-job/ops_agent.py | 279 ++++++++++++++
flink-stateful-ops-agent-demo/agent-job/ops_job.py | 111 ++++++
.../agent-job/ops_pipeline.py | 104 ++++++
.../agent-job/ops_runbook.py | 232 ++++++++++++
.../agent-job/ops_tools.py | 95 +++++
flink-stateful-ops-agent-demo/bin/demo.sh | 287 ++++++++++++++
.../bin/internal/agent_cluster.sh | 62 +++
.../bin/internal/flink_config.yaml | 114 ++++++
.../bin/internal/setup_flink.sh | 115 ++++++
flink-stateful-ops-agent-demo/docker-compose.yml | 37 ++
flink-stateful-ops-agent-demo/docs/demo-script.md | 146 ++++++++
flink-stateful-ops-agent-demo/docs/plan.md | 226 +++++++++++
.../docs/recording-runbook.md | 143 +++++++
.../platform-sim/collector.py | 86 +++++
.../platform-sim/dashboard.html | 350 +++++++++++++++++
.../platform-sim/fleet.py | 142 +++++++
.../platform-sim/platform_sim.py | 414 +++++++++++++++++++++
.../skills/balanced-task-scheduling/SKILL.md | 71 ++++
.../spikes/failover_trials.py | 83 +++++
flink-stateful-ops-agent-demo/spikes/llm_spike.py | 253 +++++++++++++
.../spikes/skew_trials.py | 51 +++
.../spikes/spike_agent.py | 120 ++++++
flink-stateful-ops-agent-demo/spikes/spike_main.py | 70 ++++
.../target-cluster/docker-compose.yml | 80 ++++
.../target-cluster/target_ctl.py | 309 +++++++++++++++
.../target-jobs/clickstream-enrichment/pom.xml | 74 ++++
.../demos/clickstream/ClickstreamEnrichment.java | 144 +++++++
28 files changed, 4209 insertions(+)
diff --git a/flink-stateful-ops-agent-demo/.gitignore b/flink-stateful-ops-agent-demo/.gitignore
new file mode 100644
index 0000000..0cdc2e1
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/.gitignore
@@ -0,0 +1,11 @@
+# Downloaded or built by bin/demo.sh setup
+flink-2.2.1/
+flink-2.2.1-bin-scala_2.12.tgz
+third_party/
+venv/
+
+# Runtime state: checkpoints, savepoints, platform-sim logs and state
+tmp/
+
+# Local session notes (starter prompts), not part of the demo
+docs/PR-*/
diff --git a/flink-stateful-ops-agent-demo/agent-job/ops_agent.py b/flink-stateful-ops-agent-demo/agent-job/ops_agent.py
new file mode 100644
index 0000000..c10703b
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/agent-job/ops_agent.py
@@ -0,0 +1,279 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""The operations agent: one instance of memory per watched job (Flink keyed state).
+
+Per window the agent receives:
+ observe -> cooldown / healthy / WAIT are decided in code; otherwise ask the model
+ decide -> the model's Decision, checked against what the runbook allows right now
+ apply -> durable redeploy through the platform API (exactly once, even across crashes)
+ escalate -> durable escalation with a report written from memory
+Every step emits a record (decision + memory snapshot) to Kafka for the dashboard.
+"""
+import functools
+import json
+import os
+import time
+from pathlib import Path
+
+from flink_agents.api.agents.agent import STRUCTURED_OUTPUT, Agent
+from flink_agents.api.agents.types import OutputSchema
+from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.decorators import action, chat_model_connection, chat_model_setup
+from flink_agents.api.events.chat_event import ChatRequestEvent, ChatResponseEvent
+from flink_agents.api.events.event import Event, InputEvent, OutputEvent
+from flink_agents.api.resource import ResourceDescriptor, ResourceName
+from flink_agents.api.runner_context import RunnerContext
+
+import ops_tools
+from ops_runbook import (
+ MAX_INCREASES,
+ PERSIST_WINDOWS,
+ Decision,
+ allowed_actions,
+ baseline_from,
+ escalation_report,
+ new_memory,
+ next_interval,
+ observation,
+ open_incident,
+ rule_decision,
+ system_prompt,
+ user_prompt,
+)
+
+RUNBOOK_PATH = "ops.runbook-path"
+COOLDOWN_SECONDS = "ops.cooldown-seconds"
+OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434")
+OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "qwen3:8b")
+MEMORY_FIELDS = ("incident", "history", "incident_seq", "tasks_reverted", "escalated", "windows")
+WINDOW_FIELDS = ("window_end", "running", "anomalous", "skewed_windows", "config", "tasks_per_tm",
+ "cpu", "task_skew", "backpressured", "records_per_sec", "recent_failovers")
+
+
+def _load(ctx: RunnerContext) -> dict:
+ mem = new_memory()
+ mem["windows"] = []
+ for field in MEMORY_FIELDS:
+ value = ctx.short_term_memory.get(field)
+ if value is not None:
+ mem[field] = value
+ return mem
+
+
+def _save(ctx: RunnerContext, mem: dict) -> None:
+ for field in MEMORY_FIELDS:
+ ctx.short_term_memory.set(field, mem.get(field))
+
+
+def _emit(ctx: RunnerContext, kind: str, window: dict, mem: dict, key: str, **extra) -> None:
+ record = {
+ "id": f"{window['job']}|{key}|{kind}",
+ "kind": kind,
+ "job": window["job"],
+ "simulated": window.get("simulated", False),
+ "at": time.time(),
+ "window": {k: window.get(k) for k in WINDOW_FIELDS},
+ "memory": {k: mem.get(k) for k in ("incident", "history", "tasks_reverted", "escalated")},
+ **extra,
+ }
+ ctx.send_event(OutputEvent(output=json.dumps(record)))
+
+
+def _close(mem: dict, outcome: str, at: float) -> None:
+ incident = mem["incident"]
+ mem["history"] = (mem.get("history") or []) + [{
+ "id": incident["id"], "outcome": outcome, "closed_at": at,
+ "baseline": incident.get("baseline"), "actions": [a["label"] for a in incident["attempts"]],
+ }]
+ mem["history"] = mem["history"][-5:]
+ if outcome == "REVERTED":
+ mem["tasks_reverted"] = True
+ mem["incident"] = None
+
+
+def _apply_decision(ctx: RunnerContext, mem: dict, window: dict, decision: Decision, **info) -> None:
+ """Turn a checked decision into memory updates and, if needed, a side effect."""
+ job, incident, config = window["job"], mem["incident"], window.get("config") or {}
+ attempts = incident["attempts"]
+ if attempts and "observed_after" not in attempts[-1]:
+ attempts[-1]["observed_after"] = observation(window)
+ target, label = None, decision.action
+ if decision.action == "APPLY_TASKS":
+ incident["baseline"] = baseline_from(mem["windows"])
+ target = {"mode": "TASKS", "interval_ms": config.get("interval_ms", 20)}
+ elif decision.action == "INCREASE_INTERVAL":
+ incident["baseline"] = incident.get("baseline") or baseline_from(mem["windows"])
+ incident["increases"] += 1
+ target = {"mode": config.get("mode"), "interval_ms": next_interval(window)}
+ label = f"INCREASE_INTERVAL {target['interval_ms']}"
+ elif decision.action == "REVERT":
+ target = incident["config_before"]
+
+ record = {"decision": decision.model_dump(), **info}
+ if target:
+ request_id = f"{job}#{incident['id']}.{len(attempts) + 1}"
+ attempts.append({"action": decision.action, "label": label, "target": target,
+ "request_id": request_id, "decided_at": window["window_end"]})
+ incident["phase"] = "applying"
+ record["request_id"] = request_id
+ ctx.send_event(Event(type="apply_config", attributes={
+ "job": job, "request_id": request_id, "mode": target["mode"],
+ "interval_ms": target["interval_ms"], "reason": f"{label} ({decision.runbook_step})",
+ }))
+ elif decision.action == "KEEP":
+ _close(mem, "KEPT", window["window_end"])
+ elif decision.action == "ESCALATE":
+ incident["phase"] = "escalating"
+ request_id = f"{job}#{incident['id']}.escalation"
+ ctx.send_event(Event(type="escalate", attributes={
+ "job": job, "request_id": request_id,
+ "title": f"FLINK-38715: unbalanced task placement persists for {job}",
+ "body": escalation_report(job, mem, window),
+ }))
+ elif decision.action == "NO_ACTION":
+ _close(mem, "NO_ACTION", window["window_end"])
+ _emit(ctx, "decision", window, mem, str(window["window_end"]), **record)
+
+
+class OpsAgent(Agent):
+ @chat_model_connection
+ @staticmethod
+ def ollama() -> ResourceDescriptor:
+ return ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_CONNECTION, base_url=OLLAMA_URL, request_timeout=180.0
+ )
+
+ @chat_model_setup
+ @staticmethod
+ def runbook_llm() -> ResourceDescriptor:
+ return ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_SETUP, connection="ollama", model=OLLAMA_MODEL,
+ temperature=0.0, num_ctx=8192, think=False, keep_alive="60m",
+ )
+
+ @action(InputEvent.EVENT_TYPE)
+ @staticmethod
+ def observe(event: Event, ctx: RunnerContext) -> None:
+ window = InputEvent.from_event(event).input
+ mem = _load(ctx)
+ mem["windows"] = (mem["windows"] + [{k: window.get(k) for k in WINDOW_FIELDS}])[-6:]
+ ctx.sensory_memory.set("window", window)
+ incident, now, key = mem["incident"], window["window_end"], str(window["window_end"])
+
+ if incident and incident.get("cooldown_until") and now < incident["cooldown_until"]:
+ _emit(ctx, "observe", window, mem, key,
+ note=f"cooldown: verify in {round(incident['cooldown_until'] - now)}s")
+ elif incident is None and not window["anomalous"]:
+ mem["escalated"] = None
+ _emit(ctx, "observe", window, mem, key, note="healthy")
+ elif incident is None and mem.get("escalated"):
+ # Handed to a human: keep watching, change nothing until the job is healthy again.
+ _emit(ctx, "observe", window, mem, key,
+ note=f"escalated as {mem['escalated']['request_id']}: hands off, waiting for a human")
+ elif incident is None and window["skewed_windows"] < PERSIST_WINDOWS:
+ _emit(ctx, "observe", window, mem, key, rule="WAIT",
+ note=f"skewed {window['skewed_windows']}/{PERSIST_WINDOWS} windows: WAIT (runbook step 1)")
+ elif incident is not None and not (window["running"] and window.get("records_per_sec")):
+ _emit(ctx, "observe", window, mem, key, note="waiting for the job to run again")
+ elif incident is not None and incident["increases"] >= MAX_INCREASES and window["anomalous"]:
+ _apply_decision(ctx, mem, window, Decision(
+ runbook_step="Step 5: still unbalanced after 3 increases",
+ rationale=f"Guardrail in code: {MAX_INCREASES} increases done, skew is still "
+ f"{window.get('task_skew')}.",
+ action="ESCALATE"), source="guardrail")
+ else:
+ if incident is None:
+ open_incident(mem, window)
+ runbook = Path(ctx.config.get_str(RUNBOOK_PATH)).read_text()
+ ctx.sensory_memory.set("asked_at", time.time())
+ ctx.send_event(ChatRequestEvent(
+ model="runbook_llm",
+ messages=[
+ ChatMessage.of(MessageRole.SYSTEM, system_prompt(runbook)),
+ ChatMessage.of(MessageRole.USER, user_prompt(window["job"], mem, window)),
+ ],
+ output_schema=OutputSchema(output_schema=Decision),
+ ))
+ _save(ctx, mem)
+
+ @action(ChatResponseEvent.EVENT_TYPE)
+ @staticmethod
+ def decide(event: Event, ctx: RunnerContext) -> None:
+ response = ChatResponseEvent.from_event(event)
+ window = ctx.sensory_memory.get("window")
+ mem = _load(ctx)
+ allowed = allowed_actions(mem, window)
+ decision, note = None, None
+ if response.is_success:
+ try:
+ parsed = response.response.extra_args.get(STRUCTURED_OUTPUT)
+ decision = parsed if isinstance(parsed, Decision) else Decision.model_validate(parsed)
+ except (ValueError, TypeError) as e:
+ note = f"unparseable model reply: {e}"[:200]
+ else:
+ note = f"model call failed: {response.error}"[:200]
+ source = "llm"
+ if decision is not None and decision.action not in allowed:
+ note = f"model chose {decision.action}; allowed now: {', '.join(allowed)}"
+ decision = None
+ if decision is None:
+ decision, source = rule_decision(mem, window), "guardrail"
+ _apply_decision(ctx, mem, window, decision, source=source, note=note, allowed=allowed,
+ llm_seconds=round(time.time() - ctx.sensory_memory.get("asked_at"), 1))
+ _save(ctx, mem)
+
+ @action("apply_config")
+ @staticmethod
+ async def apply(event: Event, ctx: RunnerContext) -> None:
+ args = [event.get_attr(k) for k in ("job", "request_id", "mode", "interval_ms", "reason")]
+ request_id = args[1]
+ receipt = await ctx.durable_execute_async(
+ ops_tools.redeploy, *args,
+ reconciler=functools.partial(ops_tools.reconcile_redeploy, *args),
+ durable_id=f"redeploy:{request_id}",
+ )
+ window = ctx.sensory_memory.get("window")
+ mem = _load(ctx)
+ incident = mem["incident"]
+ attempt = next(a for a in incident["attempts"] if a["request_id"] == request_id)
+ attempt["took_s"] = receipt.get("took_s")
+ attempt["reconciled"] = bool(receipt.get("reconciled"))
+ if attempt["action"] == "REVERT":
+ _close(mem, "REVERTED", time.time())
+ else:
+ incident["phase"] = "verifying"
+ incident["cooldown_until"] = time.time() + ctx.config.get_int(COOLDOWN_SECONDS, 30)
+ _save(ctx, mem)
+ _emit(ctx, "applied", window, mem, request_id, request_id=request_id, receipt=receipt)
+
+ @action("escalate")
+ @staticmethod
+ async def escalate(event: Event, ctx: RunnerContext) -> None:
+ args = [event.get_attr(k) for k in ("job", "request_id", "title", "body")]
+ receipt = await ctx.durable_execute_async(
+ ops_tools.file_escalation, *args,
+ reconciler=functools.partial(ops_tools.reconcile_escalation, *args),
+ durable_id=f"escalation:{args[1]}",
+ )
+ window = ctx.sensory_memory.get("window")
+ mem = _load(ctx)
+ _close(mem, "ESCALATED", time.time())
+ mem["escalated"] = {"request_id": args[1], "at": time.time()}
+ _save(ctx, mem)
+ _emit(ctx, "escalated", window, mem, args[1], request_id=args[1], receipt=receipt,
+ title=args[2], report=args[3])
diff --git a/flink-stateful-ops-agent-demo/agent-job/ops_job.py b/flink-stateful-ops-agent-demo/agent-job/ops_job.py
new file mode 100644
index 0000000..29fe52a
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/agent-job/ops_job.py
@@ -0,0 +1,111 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""Operations agent job: Kafka metrics -> event-time windows -> persistence filter -> agent -> Kafka.
+
+Submit with bin/demo.sh agent (flink run -pym ops_job -pyfs <this dir>).
+"""
+import json
+import os
+from pathlib import Path
+
+from pyflink.common import SimpleStringSchema, WatermarkStrategy
+from pyflink.common.time import Time
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.datastream.connectors import DeliveryGuarantee
+from pyflink.datastream.connectors.kafka import (
+ KafkaOffsetsInitializer,
+ KafkaRecordSerializationSchema,
+ KafkaSink,
+ KafkaSource,
+)
+from pyflink.datastream.window import TumblingEventTimeWindows
+
+from flink_agents.api.execution_environment import AgentsExecutionEnvironment
+from ops_agent import COOLDOWN_SECONDS, RUNBOOK_PATH, OpsAgent
+from ops_pipeline import PersistenceFilter, SummarizeWindow
+
+KAFKA = os.environ.get("KAFKA_BOOTSTRAP", "localhost:9092")
+METRICS_TOPIC = "ops_metrics"
+RECORDS_TOPIC = "ops_records"
+WINDOW_SECONDS = int(os.environ.get("WINDOW_SECONDS", "15"))
+COOLDOWN = int(os.environ.get("COOLDOWN_SECONDS", "30"))
+AGENT_PARALLELISM = 4
+# flink run may execute a copy of this file, so bin/demo.sh passes the absolute path.
+RUNBOOK = os.environ.get(
+ "OPS_RUNBOOK",
+ str(Path(__file__).resolve().parent.parent / "skills" / "balanced-task-scheduling" / "SKILL.md"),
+)
+
+if __name__ == "__main__":
+ env = StreamExecutionEnvironment.get_execution_environment()
+ env.set_parallelism(AGENT_PARALLELISM)
+ source = (
+ KafkaSource.builder()
+ .set_bootstrap_servers(KAFKA)
+ .set_topics(METRICS_TOPIC)
+ .set_group_id("ops-agent")
+ .set_starting_offsets(KafkaOffsetsInitializer.latest())
+ .set_value_only_deserializer(SimpleStringSchema())
+ .build()
+ )
+ samples = (
+ # Event time = the Kafka record timestamp, set by the collector to the poll time.
+ # After a crash the replayed samples land in the same windows, so the agent sees
+ # identical inputs and its action store can replay instead of re-executing.
+ env.from_source(source, WatermarkStrategy.for_monotonous_timestamps(), "metrics (Kafka)")
+ .set_parallelism(1)
+ .map(lambda s: json.loads(s))
+ .name("parse")
+ .set_parallelism(1)
+ )
+ windows = (
+ samples.key_by(lambda s: s["job"])
+ .window(TumblingEventTimeWindows.of(Time.seconds(WINDOW_SECONDS)))
+ .process(SummarizeWindow())
+ .name(f"window {WINDOW_SECONDS}s")
+ .set_parallelism(1)
+ )
+ persistent = (
+ windows.key_by(lambda w: w["job"])
+ .process(PersistenceFilter())
+ .name("persistence filter")
+ .set_parallelism(1)
+ )
+
+ agents_env = AgentsExecutionEnvironment.get_execution_environment(env)
+ agents_env.get_config().set_str(RUNBOOK_PATH, str(RUNBOOK))
+ agents_env.get_config().set_int(COOLDOWN_SECONDS, COOLDOWN)
+ records = (
+ agents_env.from_datastream(persistent, key_selector=lambda w: w["job"])
+ .apply(OpsAgent())
+ .to_datastream(output_type=Types.STRING())
+ )
+ records.sink_to(
+ KafkaSink.builder()
+ .set_bootstrap_servers(KAFKA)
+ .set_record_serializer(
+ KafkaRecordSerializationSchema.builder()
+ .set_topic(RECORDS_TOPIC)
+ .set_value_serialization_schema(SimpleStringSchema())
+ .build()
+ )
+ .set_delivery_guarantee(DeliveryGuarantee.AT_LEAST_ONCE)
+ .build()
+ ).name("records (Kafka)")
+ agents_env.execute()
diff --git a/flink-stateful-ops-agent-demo/agent-job/ops_pipeline.py b/flink-stateful-ops-agent-demo/agent-job/ops_pipeline.py
new file mode 100644
index 0000000..c369e48
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/agent-job/ops_pipeline.py
@@ -0,0 +1,104 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""Plain Flink stages in front of the agent: window the metric samples per job, then
+forward only windows worth the agent's attention."""
+from pyflink.common.typeinfo import Types
+from pyflink.datastream.functions import KeyedProcessFunction, ProcessWindowFunction
+from pyflink.datastream.state import ValueStateDescriptor
+
+# A window counts as skewed when task counts differ by 2+, the hottest TaskManager
+# is near saturation while another is well below, and upstream is backpressured.
+MIN_TASK_SKEW = 2
+HOT_CPU = 0.85
+COLD_CPU = 0.6
+BACKPRESSURE_MS = 300
+# After a config change, forward this many windows so the agent can verify it.
+FOLLOWUP_WINDOWS = 6
+
+
+class SummarizeWindow(ProcessWindowFunction):
+ """Samples of one job in one window -> placement, CPU, backpressure, throughput."""
+
+ def process(self, key, context, elements):
+ samples = sorted(elements, key=lambda s: s["ts"])
+ live = [s for s in samples if s.get("running")]
+ configs = [s["config"] for s in samples if s.get("config")]
+ summary = {
+ "job": key,
+ "simulated": bool(samples[-1].get("simulated")),
+ "window_end": context.window().end / 1000.0,
+ "samples": len(samples),
+ "running": bool(live) and len(live) == len(samples),
+ "config": configs[-1] if configs else None,
+ "recent_failovers": max(s.get("recent_failovers", 0) for s in samples),
+ "anomalous": False,
+ }
+ if live:
+ latest = live[-1]
+ tasks = latest["tasks_per_tm"]
+ cpu = {tm: round(sum(s["cpu"].get(tm, 0) for s in live) / len(live), 2) for tm in tasks}
+ counts = list(tasks.values())
+ pressured = sum(1 for s in live if s.get("backpressure_ms", 0) >= BACKPRESSURE_MS)
+ # Median of per-poll counter deltas: robust to one partial metric read.
+ same_job = [s for s in live if s.get("job_id") == latest.get("job_id")]
+ deltas = sorted(
+ (b["records_in"] - a["records_in"]) / (b["ts"] - a["ts"])
+ for a, b in zip(same_job, same_job[1:])
+ if b["ts"] - a["ts"] >= 1
+ )
+ rate = round(deltas[len(deltas) // 2]) if deltas else None
+ summary.update(
+ tasks_per_tm=tasks,
+ cpu=cpu,
+ task_skew=max(counts) - min(counts),
+ backpressured=pressured * 2 >= len(live),
+ records_per_sec=rate,
+ )
+ summary["anomalous"] = (
+ summary["task_skew"] >= MIN_TASK_SKEW
+ and max(cpu.values()) >= HOT_CPU
+ and min(cpu.values()) <= COLD_CPU
+ and summary["backpressured"]
+ )
+ yield summary
+
+
+class PersistenceFilter(KeyedProcessFunction):
+ """Drops healthy windows. Forwards skewed windows (with how many in a row), the first
+ healthy window after them, and the windows right after a config change."""
+
+ def open(self, runtime_context):
+ self.state = runtime_context.get_state(
+ ValueStateDescriptor("persistence", Types.PICKLED_BYTE_ARRAY())
+ )
+
+ def process_element(self, window, ctx):
+ st = self.state.value() or {"streak": 0, "config": None, "followups": 0, "prev_skewed": False}
+ st["streak"] = st["streak"] + 1 if window["anomalous"] else 0
+ config = window.get("config")
+ if config and st["config"] and config != st["config"]:
+ st["followups"] = FOLLOWUP_WINDOWS
+ if config:
+ st["config"] = config
+ forward = window["anomalous"] or st["prev_skewed"] or st["followups"] > 0
+ st["followups"] = max(0, st["followups"] - 1)
+ st["prev_skewed"] = window["anomalous"]
+ self.state.update(st)
+ if forward:
+ window["skewed_windows"] = st["streak"]
+ yield window
diff --git a/flink-stateful-ops-agent-demo/agent-job/ops_runbook.py b/flink-stateful-ops-agent-demo/agent-job/ops_runbook.py
new file mode 100644
index 0000000..6cefa19
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/agent-job/ops_runbook.py
@@ -0,0 +1,232 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""The runbook decision and the guardrails around it (no Flink imports).
+
+The model picks the runbook step; code decides which actions are allowed at that
+point of the incident, computes the new interval, and enforces the limits.
+"""
+import json
+from typing import Literal
+
+from pydantic import BaseModel
+
+PERSIST_WINDOWS = 3
+MAX_INCREASES = 3
+INTERVAL_STEP_MS = 50
+
+Action = Literal["WAIT", "APPLY_TASKS", "KEEP", "REVERT", "INCREASE_INTERVAL", "ESCALATE", "NO_ACTION"]
+
+
+class Decision(BaseModel):
+ runbook_step: str
+ rationale: str
+ action: Action
+ interval_ms: int | None = None
+
+
+OUTPUT_FORMAT = (
+ "Decide the single next step for this job. Reply with ONLY a JSON object, no prose, "
+ "with the fields in this order:\n"
+ '{"runbook_step": "<the matching row of the runbook decision table>", '
+ '"rationale": "<one sentence: which facts from memory and the observation match that row>", '
+ '"action": "WAIT" | "APPLY_TASKS" | "KEEP" | "REVERT" | "INCREASE_INTERVAL" | '
+ '"ESCALATE" | "NO_ACTION", "interval_ms": <new slot.request.max-interval in ms, '
+ "only for INCREASE_INTERVAL, else null>}"
+)
+
+
+def system_prompt(runbook: str) -> str:
+ return (
+ "You are an on-call operations agent for Apache Flink jobs. The memory section is "
+ "what you observed and did earlier for this job. Follow this runbook strictly:\n\n"
+ + runbook + "\n\n" + OUTPUT_FORMAT
+ )
+
+
+def new_memory() -> dict:
+ return {"incident": None, "history": [], "incident_seq": 0, "tasks_reverted": False,
+ "escalated": None}
+
+
+def observation(window: dict) -> dict:
+ """The facts from one window that the model sees."""
+ return {
+ "tasks_per_tm": window.get("tasks_per_tm"),
+ "cpu": window.get("cpu"),
+ "backpressured": window.get("backpressured"),
+ "records_per_sec": window.get("records_per_sec"),
+ "recent_failovers": window.get("recent_failovers", 0),
+ }
+
+
+def open_incident(mem: dict, window: dict) -> dict:
+ mem["incident_seq"] = mem.get("incident_seq", 0) + 1
+ incident = {
+ "id": mem["incident_seq"],
+ "opened_at": window["window_end"],
+ "config_before": window.get("config"),
+ "baseline": None,
+ "attempts": [],
+ "increases": 0,
+ "phase": "deciding",
+ "cooldown_until": None,
+ }
+ mem["incident"] = incident
+ return incident
+
+
+def baseline_from(windows: list) -> dict:
+ """Throughput and placement from the last skewed windows, before any change."""
+ skewed = [w for w in windows if w.get("anomalous")][-PERSIST_WINDOWS:]
+ rates = [w["records_per_sec"] for w in skewed if w.get("records_per_sec")]
+ last = skewed[-1] if skewed else windows[-1]
+ return {
+ "records_per_sec": round(sum(rates) / len(rates)) if rates else None,
+ "tasks_per_tm": last.get("tasks_per_tm"),
+ "cpu": last.get("cpu"),
+ "skew": last.get("task_skew"),
+ }
+
+
+def allowed_actions(mem: dict, window: dict) -> list:
+ incident = mem.get("incident") or {}
+ attempts = incident.get("attempts") or []
+ mode = (window.get("config") or {}).get("mode")
+ last = attempts[-1]["action"] if attempts else None
+ if last is None:
+ if mode == "NONE":
+ return ["ESCALATE"] if mem.get("tasks_reverted") else ["APPLY_TASKS", "WAIT", "NO_ACTION"]
+ return ["INCREASE_INTERVAL", "WAIT", "NO_ACTION"]
+ if last == "APPLY_TASKS":
+ return ["KEEP", "REVERT"]
+ if last == "INCREASE_INTERVAL":
+ if not window.get("anomalous"):
+ return ["KEEP"]
+ # Decision table: ESCALATE only once the increases are used up.
+ if incident.get("increases", 0) >= MAX_INCREASES:
+ return ["ESCALATE"]
+ return ["INCREASE_INTERVAL"]
+ return ["ESCALATE"]
+
+
+def rule_decision(mem: dict, window: dict) -> Decision:
+ """What the runbook table says, computed in code; used when the model's answer is
+ missing or not allowed at this point."""
+ allowed = allowed_actions(mem, window)
+ incident = mem.get("incident") or {}
+ baseline = (incident.get("baseline") or {}).get("records_per_sec")
+ rate = window.get("records_per_sec")
+ if "APPLY_TASKS" in allowed:
+ return Decision(runbook_step="Step 2", rationale="Skew persisted, TASKS not tried yet.",
+ action="APPLY_TASKS")
+ if allowed == ["KEEP", "REVERT"]:
+ worse = baseline and rate is not None and rate < baseline
+ return Decision(runbook_step="Step 3", rationale="Compared with the baseline.",
+ action="REVERT" if worse else "KEEP")
+ if "INCREASE_INTERVAL" in allowed:
+ return Decision(runbook_step="Step 4", rationale="TASKS on and the skew is back.",
+ action="INCREASE_INTERVAL")
+ return Decision(runbook_step="Step 5" if allowed == ["ESCALATE"] else "Step 3",
+ rationale="Only allowed action at this point.", action=allowed[0])
+
+
+def prompt_memory(mem: dict, window: dict) -> dict:
+ """Memory as the model sees it: the facts that select a row of the decision table."""
+ incident = mem.get("incident") or {}
+ attempts = incident.get("attempts") or []
+ view = {
+ "incident": "open",
+ "windows_with_skew": window.get("skewed_windows", 0),
+ "config": window.get("config"),
+ "baseline": incident.get("baseline"),
+ "actions": [a["label"] for a in attempts],
+ }
+ if attempts:
+ view["cooldown_over"] = True
+ if not attempts and (window.get("config") or {}).get("mode") == "TASKS":
+ view["since_last_action"] = (
+ f"TASKS was already on; {window.get('recent_failovers', 0)} failovers; "
+ f"skew back for {window.get('skewed_windows', 0)} windows"
+ )
+ if attempts and attempts[-1]["action"] == "INCREASE_INTERVAL":
+ view["increases_so_far"] = incident.get("increases", 0)
+ view["increases_allowed"] = MAX_INCREASES
+ view["after_each_increase"] = (
+ f"still skew {window.get('task_skew')} after redeploy" if window.get("anomalous")
+ else "balanced after redeploy"
+ )
+ return view
+
+
+def user_prompt(job: str, mem: dict, window: dict) -> str:
+ return (
+ f"Job: {job}\nMemory: {json.dumps(prompt_memory(mem, window))}\n"
+ f"Latest observation: {json.dumps(observation(window))}\n"
+ f"Allowed actions now: {', '.join(allowed_actions(mem, window))}"
+ )
+
+
+def next_interval(window: dict) -> int:
+ return int((window.get("config") or {}).get("interval_ms") or 20) + INTERVAL_STEP_MS
+
+
+def _cfg(config: dict | None) -> str:
+ if not config:
+ return "n/a"
+ return f"mode={config.get('mode')}, max-interval={config.get('interval_ms')}ms"
+
+
+def _per_tm(values: dict | None, pct: bool = False) -> str:
+ if not values:
+ return "n/a"
+ return " / ".join(f"{tm} {round(v * 100)}%" if pct else f"{tm}: {v}"
+ for tm, v in sorted(values.items()))
+
+
+def escalation_report(job: str, mem: dict, window: dict) -> str:
+ """The FLINK-38715 comment, written from the agent's memory."""
+ incident = mem["incident"]
+ base = incident.get("baseline") or {}
+ lines = [
+ f"### Unbalanced task placement persists: {job}",
+ "",
+ "Runbook: balanced-task-scheduling (Flink 2.2, FLIP-370), steps 1-5 followed by "
+ "the Flink Agents operations agent.",
+ "",
+ f"- Config before: `{_cfg(incident.get('config_before'))}`",
+ f"- Config now: `{_cfg(window.get('config'))}`",
+ f"- Baseline: {base.get('records_per_sec') or 'n/a'} rec/s, tasks per TM "
+ f"`{_per_tm(base.get('tasks_per_tm'))}`" if base else "- Baseline: not recorded",
+ f"- Recent failovers: {window.get('recent_failovers', 0)}",
+ "",
+ "| # | Action | Config after | Tasks per TM after | Rec/s after |",
+ "|---|---|---|---|---|",
+ ]
+ for i, a in enumerate(incident.get("attempts") or [], 1):
+ after = a.get("observed_after") or {}
+ lines.append(
+ f"| {i} | {a['label']} | {_cfg(a.get('target'))} | "
+ f"{_per_tm(after.get('tasks_per_tm'))} | {after.get('records_per_sec', 'n/a')} |"
+ )
+ lines += [
+ "",
+ f"Latest: tasks per TM `{_per_tm(window.get('tasks_per_tm'))}`, CPU "
+ f"`{_per_tm(window.get('cpu'), pct=True)}`, {window.get('records_per_sec')} rec/s.",
+ f"Stopped after {incident.get('increases', 0)} interval increases (limit {MAX_INCREASES}).",
+ ]
+ return "\n".join(lines)
diff --git a/flink-stateful-ops-agent-demo/agent-job/ops_tools.py b/flink-stateful-ops-agent-demo/agent-job/ops_tools.py
new file mode 100644
index 0000000..ab3086c
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/agent-job/ops_tools.py
@@ -0,0 +1,95 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""Side effects of the agent, called through durable execution.
+
+Each function talks to the platform's deployment API (platform-sim stands in for
+it). The platform executes every request it receives; it does not de-duplicate.
+Exactly-once comes from the agent: a completed call is replayed from the action
+store on recovery, and an in-flight call goes through the reconciler, which asks
+the platform whether the request already exists instead of sending it again.
+"""
+import json
+import os
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+PLATFORM_URL = os.environ.get("OPS_PLATFORM_URL", "http://localhost:8090")
+DEPLOY_TIMEOUT_S = 300
+
+
+def _call(method, path, body=None):
+ data = json.dumps(body).encode() if body is not None else None
+ req = urllib.request.Request(
+ PLATFORM_URL + path, data=data, method=method, headers={"Content-Type": "application/json"}
+ )
+ try:
+ with urllib.request.urlopen(req, timeout=30) as resp:
+ return json.loads(resp.read() or b"{}")
+ except urllib.error.HTTPError as e:
+ if e.code == 404:
+ return None
+ raise
+
+
+def _id(request_id):
+ return urllib.parse.quote(request_id, safe="")
+
+
+def _wait_done(request_id):
+ deadline = time.time() + DEPLOY_TIMEOUT_S
+ while time.time() < deadline:
+ dep = _call("GET", f"/api/deployments/{_id(request_id)}")
+ if dep and dep["status"] == "done":
+ return {k: dep.get(k) for k in ("request_id", "job", "target", "result", "took_s")}
+ if dep and dep["status"] == "failed":
+ raise RuntimeError(f"redeploy {request_id} failed: {dep.get('error')}")
+ time.sleep(1)
+ raise TimeoutError(f"redeploy {request_id} did not finish in {DEPLOY_TIMEOUT_S}s")
+
+
+def redeploy(job, request_id, mode, interval_ms, reason):
+ """Ask the platform to apply a scheduling config (savepoint, recreate, restore)."""
+ _call("POST", "/api/deployments", {
+ "request_id": request_id, "job": job, "mode": mode, "interval_ms": interval_ms,
+ "reason": reason, "agent_pid": os.getpid(),
+ })
+ return _wait_done(request_id)
+
+
+def reconcile_redeploy(job, request_id, mode, interval_ms, reason):
+ """Recovery only: the call was in flight when the agent crashed. Did it reach the platform?"""
+ if _call("GET", f"/api/deployments/{_id(request_id)}") is None:
+ return redeploy(job, request_id, mode, interval_ms, reason)
+ _call("POST", f"/api/deployments/{_id(request_id)}/reconciled", {"agent_pid": os.getpid()})
+ receipt = _wait_done(request_id)
+ receipt["reconciled"] = True
+ return receipt
+
+
+def file_escalation(job, request_id, title, body):
+ """Post the escalation (in production: a JIRA comment or a page)."""
+ _call("POST", "/api/escalations", {"request_id": request_id, "job": job, "title": title, "body": body})
+ return {"request_id": request_id, "filed": True}
+
+
+def reconcile_escalation(job, request_id, title, body):
+ if _call("GET", f"/api/escalations/{_id(request_id)}") is None:
+ return file_escalation(job, request_id, title, body)
+ return {"request_id": request_id, "filed": True, "reconciled": True}
diff --git a/flink-stateful-ops-agent-demo/bin/demo.sh b/flink-stateful-ops-agent-demo/bin/demo.sh
new file mode 100755
index 0000000..8fd4b89
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/bin/demo.sh
@@ -0,0 +1,287 @@
+#!/usr/bin/env bash
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+#
+# Stateful operations agent demo. See docs/recording-runbook.md.
+#
+# setup one time: agent Flink, venv, flink-agents (main + #1161), target job jar, model
+# up Kafka + topics, target cluster (NONE, skewed) + ClickstreamEnrichment,
+# agent cluster (2 TaskManagers), Ollama warm-up, platform-sim (background)
+# platform run platform-sim in the foreground instead (logs in the terminal)
+# agent submit the operations agent job
+# fleet-incident start the SIMULATED fleet incident (fleet/payments-sessionizer)
+# kill-agent-tm [--wait]
+# kill -9 the agent TaskManager that sent the in-flight redeploy;
+# with --wait, first wait for a redeploy to start (then 5s more)
+# reset back to the starting point for another take
+# status what is running
+# down [--all] stop the clusters and platform-sim (--all: Kafka too, unload the model)
+#
+# Environment: TM_CPUS (target TaskManager CPUs, default 0.5: verified for recording, easy on a laptop),
+# TARGET_RATE (target job input, records/s; default 19000 × TM_CPUS, i.e. 9500),
+# WINDOW_SECONDS (default 15), COOLDOWN_SECONDS (default 30), KILL_DELAY (default 5).
+set -euo pipefail
+
+PROJECT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+FLINK_HOME="$PROJECT_DIR/flink-2.2.1"
+TMP="$PROJECT_DIR/tmp"
+PLATFORM="http://localhost:8090"
+AGENT_REST="http://localhost:8082"
+TARGET_REST="http://localhost:8081"
+OLLAMA="http://localhost:11434"
+MODEL="${OLLAMA_MODEL:-qwen3:8b}"
+KAFKA_CONTAINER="stateful-ops-kafka"
+export TM_CPUS="${TM_CPUS:-0.5}"
+export WINDOW_SECONDS="${WINDOW_SECONDS:-15}"
+export COOLDOWN_SECONDS="${COOLDOWN_SECONDS:-30}"
+export OPS_RUNBOOK="$PROJECT_DIR/skills/balanced-task-scheduling/SKILL.md"
+
+say() { printf '\033[1m==> %s\033[0m\n' "$*"; }
+die() { printf '\033[31m%s\033[0m\n' "$*" >&2; exit 1; }
+venv() {
+ # shellcheck disable=SC1091
+ source "$PROJECT_DIR/venv/bin/activate"
+ PYTHONPATH="$(python -c 'import sysconfig; print(sysconfig.get_paths()["purelib"])')"
+ export PYTHONPATH
+}
+py() { "$PROJECT_DIR/venv/bin/python" "$@"; }
+target_ctl() { py "$PROJECT_DIR/target-cluster/target_ctl.py" "$@"; }
+up_ok() { curl -sf -o /dev/null "$1"; }
+
+kafka_topics() { docker exec "$KAFKA_CONTAINER" /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 "$@"; }
+
+start_kafka() {
+ docker compose -f "$PROJECT_DIR/docker-compose.yml" up -d kafka >/dev/null 2>&1
+ for _ in $(seq 1 30); do kafka_topics --list >/dev/null 2>&1 && break; sleep 2; done
+ kafka_topics --create --if-not-exists --topic ops_metrics --partitions 1 >/dev/null
+ kafka_topics --create --if-not-exists --topic ops_records --partitions 1 >/dev/null
+ kafka_topics --create --if-not-exists --topic agent_action_state --partitions 8 >/dev/null
+ say "Kafka up (ops_metrics, ops_records, agent_action_state)"
+}
+
+# Is ClickstreamEnrichment running in NONE mode with task skew >= 2? Waits up to 30s for metrics.
+target_skewed() {
+ PYTHONPATH="$PROJECT_DIR/target-cluster:$PROJECT_DIR/platform-sim" py - <<'EOF'
+import sys, time
+import collector
+for _ in range(15):
+ s = collector.sample("t", time.time())
+ if s.get("running"):
+ counts = list(s["tasks_per_tm"].values())
+ print(f" target: mode {s['config']['mode']}, tasks per TM {s['tasks_per_tm']}, cpu {s['cpu']}")
+ sys.exit(0 if s["config"]["mode"] == "NONE" and max(counts) - min(counts) >= 2 else 1)
+ time.sleep(2)
+sys.exit(1)
+EOF
+}
+
+# Target cluster in NONE mode with a skewed placement. NONE placement is not
+# guaranteed to skew, so recreate (up to 3 times) until it does.
+ensure_target() {
+ local fresh="${1:-}"
+ for attempt in 1 2 3; do
+ if [ -n "$fresh" ] || ! target_skewed >/dev/null 2>&1; then
+ say "Target cluster: NONE mode, TM_CPUS=$TM_CPUS, submit ClickstreamEnrichment (attempt $attempt)"
+ target_ctl up --mode NONE --interval 20 >/dev/null
+ target_ctl submit >/dev/null
+ fi
+ if target_skewed; then say "Target skewed: ready ($TARGET_REST)"; return 0; fi
+ fresh=1
+ done
+ die "Target placement did not skew after 3 attempts; run: bin/demo.sh reset"
+}
+
+# A freshly started target needs ~30-60s (JIT warm-up) before the skewed placement makes it
+# fall behind its input rate. Wait for that, so a take starts with backpressure on screen.
+wait_target_backpressured() {
+ say "Waiting for the target to fall behind its input (backpressure)..."
+ PYTHONPATH="$PROJECT_DIR/target-cluster:$PROJECT_DIR/platform-sim" py - <<'EOF' || say "warning: target not backpressured yet; the first skewed window will come later"
+import sys, time
+import collector
+streak = 0
+for _ in range(60):
+ s = collector.sample("t", time.time())
+ bp = s.get("backpressure_ms") or 0
+ streak = streak + 1 if s.get("running") and bp >= 300 else 0
+ if streak >= 3:
+ print(f" target backpressured: Enrich {bp} ms/s")
+ sys.exit(0)
+ time.sleep(2)
+sys.exit(1)
+EOF
+}
+
+configure_agent_cluster() {
+ local conf="$FLINK_HOME/conf/config.yaml"
+ mkdir -p "$TMP/checkpoints" "$TMP/savepoints"
+ cp "$PROJECT_DIR/bin/internal/flink_config.yaml" "$conf"
+ sed -i '' -e "s|CHECKPOINT_DIR_PLACEHOLDER|$TMP/checkpoints|" -e "s|SAVEPOINT_DIR_PLACEHOLDER|$TMP/savepoints|" \
+ -e "s|PYTHON_EXECUTABLE_PLACEHOLDER|$PROJECT_DIR/venv/bin/python|g" "$conf"
+}
+
+start_agent_cluster() {
+ if up_ok "$AGENT_REST/overview"; then
+ say "Agent cluster already up ($AGENT_REST)"
+ return
+ fi
+ configure_agent_cluster
+ "$PROJECT_DIR/bin/internal/agent_cluster.sh" start
+}
+
+restart_agent_cluster() {
+ "$PROJECT_DIR/bin/internal/agent_cluster.sh" stop >/dev/null 2>&1 || true
+ for _ in $(seq 1 15); do up_ok "$AGENT_REST/overview" || break; sleep 1; done
+ configure_agent_cluster
+ "$PROJECT_DIR/bin/internal/agent_cluster.sh" start
+}
+
+warm_model() {
+ up_ok "$OLLAMA/api/tags" || die "Ollama is not running: open the Ollama app (or run: ollama serve)"
+ curl -sf "$OLLAMA/api/generate" -d "{\"model\":\"$MODEL\",\"prompt\":\"ok\",\"stream\":false,\"think\":false,\"keep_alive\":\"60m\",\"options\":{\"num_predict\":1,\"num_ctx\":8192}}" >/dev/null \
+ || die "Could not load $MODEL (ollama pull $MODEL)"
+ say "Model $MODEL loaded (kept for 60m)"
+}
+
+start_platform() {
+ if up_ok "$PLATFORM/api/state"; then
+ say "platform-sim already up ($PLATFORM)"
+ return
+ fi
+ mkdir -p "$TMP/platform"
+ nohup "$PROJECT_DIR/venv/bin/python" "$PROJECT_DIR/platform-sim/platform_sim.py" >>"$TMP/platform/platform.log" 2>&1 &
+ echo $! >"$TMP/platform/platform.pid"
+ for _ in $(seq 1 20); do up_ok "$PLATFORM/api/state" && break; sleep 0.5; done
+ up_ok "$PLATFORM/api/state" || die "platform-sim did not start; see $TMP/platform/platform.log"
+ say "platform-sim up: dashboard $PLATFORM (log: tmp/platform/platform.log)"
+}
+
+stop_platform() {
+ if [ -f "$TMP/platform/platform.pid" ]; then
+ kill "$(cat "$TMP/platform/platform.pid")" 2>/dev/null || true
+ rm -f "$TMP/platform/platform.pid"
+ fi
+}
+
+running_agent_jobs() {
+ curl -sf "$AGENT_REST/jobs/overview" 2>/dev/null | py -c '
+import json, sys
+for j in json.load(sys.stdin)["jobs"]:
+ if j["state"] not in ("FINISHED", "CANCELED", "FAILED"):
+ print(j["jid"])' || true
+}
+
+cancel_agent_jobs() {
+ for jid in $(running_agent_jobs); do
+ curl -sf -X PATCH "$AGENT_REST/jobs/$jid?mode=cancel" >/dev/null || true
+ say "Cancelled agent job $jid"
+ done
+}
+
+cmd_setup() {
+ "$PROJECT_DIR/bin/internal/setup_flink.sh"
+ if [ ! -f "$PROJECT_DIR/target-jobs/clickstream-enrichment/target/clickstream-enrichment.jar" ]; then
+ (cd "$PROJECT_DIR/target-jobs/clickstream-enrichment" && mvn -q -B package)
+ fi
+ command -v ollama >/dev/null && ollama pull "$MODEL"
+ say "Setup done. Next: bin/demo.sh up"
+}
+
+cmd_up() {
+ start_kafka
+ ensure_target
+ start_agent_cluster
+ warm_model
+ start_platform
+ wait_target_backpressured
+ say "Ready. Dashboard $PLATFORM · target UI $TARGET_REST · agent UI $AGENT_REST. Next: bin/demo.sh agent"
+}
+
+cmd_agent() {
+ [ -z "$(running_agent_jobs)" ] || die "An agent job is already running; run: bin/demo.sh reset"
+ up_ok "$PLATFORM/api/state" || die "platform-sim is not running; run: bin/demo.sh up"
+ venv
+ say "Submitting the operations agent (windows ${WINDOW_SECONDS}s, cooldown ${COOLDOWN_SECONDS}s)"
+ "$FLINK_HOME/bin/flink" run -d -Dpipeline.name=stateful-ops-agent -pym ops_job -pyfs "$PROJECT_DIR/agent-job"
+ say "Agent job submitted: $AGENT_REST"
+}
+
+cmd_kill_agent_tm() {
+ local pid="" job="" rid=""
+ if [ "${1:-}" = "--wait" ]; then
+ say "Waiting for the agent to start a redeploy..."
+ until curl -sf "$PLATFORM/api/inflight" >/dev/null; do sleep 0.5; done
+ sleep "${KILL_DELAY:-5}"
+ fi
+ read -r pid job rid < <(curl -sf "$PLATFORM/api/inflight" | py -c '
+import json, sys
+d = json.load(sys.stdin)
+print(d.get("agent_pid") or "", d["job"], d["request_id"])') || die "No redeploy in flight"
+ ps -p "$pid" -o command= 2>/dev/null | grep -q TaskManagerRunner || die "pid $pid is not a TaskManager"
+ echo "Redeploy in flight: $rid ($job), sent by the agent TaskManager pid $pid"
+ echo "\$ kill -9 $pid"
+ kill -9 "$pid"
+}
+
+cmd_reset() {
+ cancel_agent_jobs
+ start_platform
+ curl -sf -X POST "$PLATFORM/api/reset" >/dev/null && say "platform-sim reset"
+ ensure_target fresh
+ say "Restarting the agent cluster (fresh Python workers, 2 TaskManagers)"
+ restart_agent_cluster
+ warm_model
+ wait_target_backpressured
+ say "Ready for a take. Next: bin/demo.sh agent"
+}
+
+cmd_status() {
+ up_ok "$PLATFORM/api/state" && echo "platform-sim: up ($PLATFORM)" || echo "platform-sim: down"
+ docker ps --format '{{.Names}}: {{.Status}}' | grep -E "stateful-ops|target" || echo "no containers"
+ if up_ok "$AGENT_REST/overview"; then
+ curl -sf "$AGENT_REST/overview" | py -c 'import json,sys; o=json.load(sys.stdin); print("agent cluster: {taskmanagers} TMs, {slots-available}/{slots-total} slots free, jobs running {jobs-running}".format_map(o))'
+ else
+ echo "agent cluster: down"
+ fi
+ target_skewed 2>/dev/null || echo " (target not skewed / not running)"
+ curl -sf "$OLLAMA/api/ps" | py -c 'import json,sys; print("ollama loaded:", [m["name"] for m in json.load(sys.stdin)["models"]])' 2>/dev/null || echo "ollama: down"
+}
+
+cmd_down() {
+ cancel_agent_jobs
+ stop_platform
+ "$PROJECT_DIR/bin/internal/agent_cluster.sh" stop >/dev/null 2>&1 || true
+ target_ctl down >/dev/null 2>&1 || true
+ if [ "${1:-}" = "--all" ]; then
+ docker compose -f "$PROJECT_DIR/docker-compose.yml" down >/dev/null 2>&1 || true
+ curl -sf "$OLLAMA/api/generate" -d "{\"model\":\"$MODEL\",\"keep_alive\":0}" >/dev/null 2>&1 || true
+ fi
+ say "Stopped"
+}
+
+case "${1:-}" in
+ setup) cmd_setup ;;
+ up) cmd_up ;;
+ platform) stop_platform; mkdir -p "$TMP/platform"; exec "$PROJECT_DIR/venv/bin/python" "$PROJECT_DIR/platform-sim/platform_sim.py" ;;
+ agent) cmd_agent ;;
+ fleet-incident) curl -sf -X POST "$PLATFORM/api/fleet/incident" && echo ;;
+ kill-agent-tm) shift; cmd_kill_agent_tm "${1:-}" ;;
+ reset) cmd_reset ;;
+ status) cmd_status ;;
+ down) shift; cmd_down "${1:-}" ;;
+ *) sed -n '20,37p' "$0"; exit 1 ;;
+esac
diff --git a/flink-stateful-ops-agent-demo/bin/internal/agent_cluster.sh b/flink-stateful-ops-agent-demo/bin/internal/agent_cluster.sh
new file mode 100755
index 0000000..bf17535
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/bin/internal/agent_cluster.sh
@@ -0,0 +1,62 @@
+#!/bin/bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+# Start/stop the AGENT Flink cluster: 1 JobManager + NUM_TASK_MANAGERS TaskManagers.
+# Usage: agent_cluster.sh start|add-tm|stop
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_DIR="$(dirname "$(dirname "$SCRIPT_DIR")")"
+FLINK_HOME="${FLINK_HOME:-$PROJECT_DIR/flink-2.2.1}"
+NUM_TASK_MANAGERS="${NUM_TASK_MANAGERS:-2}"
+
+case "$1" in
+ start)
+ source "$PROJECT_DIR/venv/bin/activate"
+ # The embedded interpreter (Pemja) in the TaskManagers needs the venv packages.
+ export PYTHONPATH=$(python -c 'import sysconfig; print(sysconfig.get_paths()["purelib"])')
+ "$FLINK_HOME/bin/jobmanager.sh" start
+ for _ in $(seq 1 "$NUM_TASK_MANAGERS"); do
+ "$FLINK_HOME/bin/taskmanager.sh" start
+ done
+ for _ in $(seq 1 30); do
+ if curl -sf http://localhost:8082/overview | grep -q "\"taskmanagers\":$NUM_TASK_MANAGERS"; then
+ echo "Agent cluster ready: http://localhost:8082"
+ exit 0
+ fi
+ sleep 2
+ done
+ echo "Agent cluster did not become ready in time" >&2
+ exit 1
+ ;;
+ add-tm)
+ # Bring a TaskManager back after a chaos kill.
+ source "$PROJECT_DIR/venv/bin/activate"
+ export PYTHONPATH=$(python -c 'import sysconfig; print(sysconfig.get_paths()["purelib"])')
+ "$FLINK_HOME/bin/taskmanager.sh" start
+ ;;
+ stop)
+ "$FLINK_HOME/bin/taskmanager.sh" stop-all
+ "$FLINK_HOME/bin/jobmanager.sh" stop-all
+ ;;
+ *)
+ echo "Usage: $0 start|add-tm|stop" >&2
+ exit 1
+ ;;
+esac
diff --git a/flink-stateful-ops-agent-demo/bin/internal/flink_config.yaml b/flink-stateful-ops-agent-demo/bin/internal/flink_config.yaml
new file mode 100644
index 0000000..9730fdf
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/bin/internal/flink_config.yaml
@@ -0,0 +1,114 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+# Flink configuration for the AGENT cluster (the cluster that runs the
+# operations agent). The monitored "target" cluster runs in Docker.
+
+env:
+ java:
+ opts:
+ all: --add-exports=java.base/sun.net.util=ALL-UNNAMED --add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-exports=java.security.jgss/sun.security.krb5=ALL-UNNA [...]
+
+jobmanager:
+ bind-host: localhost
+ rpc:
+ address: localhost
+ port: 6123
+ memory:
+ process:
+ size: 1600m
+ execution:
+ failover-strategy: region
+
+# Two TaskManagers with 4 slots each: the agent runs at parallelism 4, so
+# killing one TaskManager still leaves enough slots to fail over.
+taskmanager:
+ bind-host: localhost
+ host: localhost
+ numberOfTaskSlots: 4
+ memory:
+ process:
+ size: 2048m
+
+parallelism:
+ default: 4
+
+# Notice a killed TaskManager within seconds (the demo's crash scene).
+heartbeat:
+ interval: 2000
+ timeout: 10000
+
+rest:
+ address: localhost
+ bind-address: localhost
+ port: 8082
+
+web:
+ submit:
+ enable: true
+ cancel:
+ enable: true
+
+# Agent memory (short-term memory) is Flink keyed state, so it is
+# checkpointed together with the stream position.
+state:
+ backend:
+ type: hashmap
+
+execution:
+ checkpointing:
+ interval: 10s
+ dir: file://CHECKPOINT_DIR_PLACEHOLDER
+ savepoint-dir: file://SAVEPOINT_DIR_PLACEHOLDER
+ externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
+
+restart-strategy:
+ type: fixed-delay
+ fixed-delay:
+ attempts: 100
+ delay: 5s
+
+# pemja: `flink run -py` also ships opt/flink-python as a user jar. Without this,
+# every job restart loads pemja (and libpython) in a fresh user classloader and
+# fails with "Native Library ... already loaded in another classloader", i.e.
+# the agent job could never fail over.
+classloader.parent-first-patterns.additional: jakarta.json;org.eclipse.parsson;pemja
+
+# Python workers use the module's venv regardless of the shell that started
+# the TaskManagers.
+python:
+ executable: PYTHON_EXECUTABLE_PLACEHOLDER
+ client:
+ executable: PYTHON_EXECUTABLE_PLACEHOLDER
+
+agent:
+ # Durable execution: completed actions / durable_execute results are
+ # recorded in Kafka, so recovery never repeats an LLM call or a side effect.
+ actionStateStoreBackend: kafka
+ kafkaBootstrapServers: localhost:9092
+ kafkaActionStateTopic: agent_action_state
+ kafkaActionStateTopicNumPartitions: 8
+ # Stable identity across restarts/resubmissions of the agent job.
+ job-identifier: stateful-ops-agent
+ # A slow or failing LLM call is retried instead of failing the job.
+ error-handling-strategy: retry
+ max-retries: 3
+ retry-wait-interval: 5
+ # Event log as audit trail (SLF4J -> visible in the Flink UI "Logs" tab).
+ eventLoggerType: slf4j
+ event-log.level: STANDARD
diff --git a/flink-stateful-ops-agent-demo/bin/internal/setup_flink.sh b/flink-stateful-ops-agent-demo/bin/internal/setup_flink.sh
new file mode 100755
index 0000000..3c3ad03
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/bin/internal/setup_flink.sh
@@ -0,0 +1,115 @@
+#!/bin/bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+# Sets up the AGENT Flink cluster (Flink 2.2.1 + Flink Agents 0.3.1) and the
+# Python virtual environment. Does not start anything.
+
+set -e
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_DIR="$(dirname "$(dirname "$SCRIPT_DIR")")"
+
+FLINK_VERSION=2.2.1
+KAFKA_CONNECTOR_VERSION=5.0.0-2.2
+# Durable execution across a TaskManager crash needs apache/flink-agents#1161
+# (not released yet), so flink-agents is built from source: main + that PR.
+FLINK_AGENTS_REPO="${FLINK_AGENTS_REPO:-https://github.com/apache/flink-agents.git}"
+FLINK_AGENTS_REF="${FLINK_AGENTS_REF:-pull/1161/head}"
+FLINK_AGENTS_SRC="$PROJECT_DIR/third_party/flink-agents"
+
+cd "$PROJECT_DIR"
+
+if [ ! -d "flink-$FLINK_VERSION" ]; then
+ if [ ! -f "flink-$FLINK_VERSION-bin-scala_2.12.tgz" ]; then
+ curl -LO "https://archive.apache.org/dist/flink/flink-$FLINK_VERSION/flink-$FLINK_VERSION-bin-scala_2.12.tgz"
+ fi
+ tar -xzf "flink-$FLINK_VERSION-bin-scala_2.12.tgz"
+fi
+export FLINK_HOME="$PROJECT_DIR/flink-$FLINK_VERSION"
+
+cp "$FLINK_HOME/opt/flink-python-$FLINK_VERSION.jar" "$FLINK_HOME/lib/"
+
+KAFKA_JAR="flink-sql-connector-kafka-$KAFKA_CONNECTOR_VERSION.jar"
+if [ ! -f "$FLINK_HOME/lib/$KAFKA_JAR" ]; then
+ curl -L -o "$FLINK_HOME/lib/$KAFKA_JAR" \
+ "https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/$KAFKA_CONNECTOR_VERSION/$KAFKA_JAR"
+fi
+
+mkdir -p "$PROJECT_DIR/tmp/checkpoints" "$PROJECT_DIR/tmp/savepoints"
+cp "$SCRIPT_DIR/flink_config.yaml" "$FLINK_HOME/conf/config.yaml"
+if [[ "$OSTYPE" == "darwin"* ]]; then SED_I=(sed -i ''); else SED_I=(sed -i); fi
+"${SED_I[@]}" "s|CHECKPOINT_DIR_PLACEHOLDER|$PROJECT_DIR/tmp/checkpoints|" "$FLINK_HOME/conf/config.yaml"
+"${SED_I[@]}" "s|SAVEPOINT_DIR_PLACEHOLDER|$PROJECT_DIR/tmp/savepoints|" "$FLINK_HOME/conf/config.yaml"
+"${SED_I[@]}" "s|PYTHON_EXECUTABLE_PLACEHOLDER|$PROJECT_DIR/venv/bin/python|g" "$FLINK_HOME/conf/config.yaml"
+
+if [ ! -d "venv" ]; then
+ PYTHON_BIN=""
+ for candidate in python3.11 python3.12 python3.10 python3; do
+ if command -v "$candidate" >/dev/null 2>&1 && \
+ "$candidate" -c 'import sys; sys.exit(0 if (3,10) <= sys.version_info[:2] <= (3,12) else 1)'; then
+ PYTHON_BIN="$candidate"
+ break
+ fi
+ done
+ if [ -z "$PYTHON_BIN" ]; then
+ echo "Error: Python 3.10-3.12 is required"
+ exit 1
+ fi
+ "$PYTHON_BIN" -m venv venv
+fi
+source venv/bin/activate
+BUILD_CONSTRAINT_FILE="$(mktemp)"
+echo "setuptools<81" > "$BUILD_CONSTRAINT_FILE"
+PIP_CONSTRAINT="$BUILD_CONSTRAINT_FILE" pip install -q \
+ "apache-flink==$FLINK_VERSION" "kafka-python>=2.0" "setuptools>=75.3,<82"
+rm -f "$BUILD_CONSTRAINT_FILE"
+
+if [ ! -d "$FLINK_AGENTS_SRC" ]; then
+ git clone -q "$FLINK_AGENTS_REPO" "$FLINK_AGENTS_SRC"
+ git -C "$FLINK_AGENTS_SRC" fetch -q origin "$FLINK_AGENTS_REF:demo-build"
+ git -C "$FLINK_AGENTS_SRC" checkout -q demo-build
+fi
+if [ "${REBUILD_FLINK_AGENTS:-0}" = "1" ] || \
+ ! ls "$FLINK_AGENTS_SRC"/dist/flink-2.2/target/flink-agents-dist-flink-2.2-*-thin.jar >/dev/null 2>&1; then
+ echo "Building flink-agents from $FLINK_AGENTS_SRC (JDK 17, a few minutes)..."
+ # `clean` matters: an incremental build can re-shade a stale runtime jar.
+ (cd "$FLINK_AGENTS_SRC" && JAVA_HOME="${JAVA17_HOME:-$(/usr/libexec/java_home -v 17)}" \
+ mvn clean install -B -q -DskipTests -Dspotless.skip=true -Drat.skip=true \
+ -pl dist/common,dist/flink-2.2 -am)
+fi
+JAR_DIR="$FLINK_AGENTS_SRC/python/flink_agents/lib"
+rm -rf "$JAR_DIR/common" "$JAR_DIR/flink-2.2"
+mkdir -p "$JAR_DIR/common" "$JAR_DIR/flink-2.2"
+cp "$FLINK_AGENTS_SRC"/dist/common/target/flink-agents-dist-common-*-SNAPSHOT.jar "$JAR_DIR/common/"
+cp "$FLINK_AGENTS_SRC"/dist/flink-2.2/target/flink-agents-dist-flink-2.2-*-thin.jar "$JAR_DIR/flink-2.2/"
+export FLINK_AGENTS_SKIP_JAR_DOWNLOAD=1
+pip install -q "$FLINK_AGENTS_SRC/python"
+pip install -q --force-reinstall --no-deps "$FLINK_AGENTS_SRC/python"
+unset FLINK_AGENTS_SKIP_JAR_DOWNLOAD
+
+PURELIB=$(python -c 'import sysconfig; print(sysconfig.get_paths()["purelib"])')
+# Python jobs: AgentsExecutionEnvironment registers the Flink Agents dist jars via
+# pipeline.jars. Do NOT also copy them into $FLINK_HOME/lib, otherwise
+# kafka-clients is loaded twice and the Kafka action state store fails with a
+# ClassCastException (JmxReporter).
+rm -f "$FLINK_HOME"/lib/flink-agents-dist-*.jar
+echo "Flink Agents dist jars are provided by $PURELIB/flink_agents/lib (pipeline.jars)"
+
+echo "Agent Flink cluster set up at $FLINK_HOME"
diff --git a/flink-stateful-ops-agent-demo/docker-compose.yml b/flink-stateful-ops-agent-demo/docker-compose.yml
new file mode 100644
index 0000000..d09a544
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/docker-compose.yml
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Shared infrastructure: Kafka (agent input/output topics + the Flink Agents
+# action state store).
+services:
+ kafka:
+ image: apache/kafka:3.9.0
+ container_name: stateful-ops-kafka
+ ports:
+ - "9092:9092"
+ environment:
+ KAFKA_NODE_ID: 1
+ KAFKA_PROCESS_ROLES: broker,controller
+ KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://:9093
+ KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
+ KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
+ KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+ KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+ KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+ KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+ KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+ KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
diff --git a/flink-stateful-ops-agent-demo/docs/demo-script.md b/flink-stateful-ops-agent-demo/docs/demo-script.md
new file mode 100644
index 0000000..44c9b34
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/docs/demo-script.md
@@ -0,0 +1,146 @@
+# Demo script: live narration over a muted video (OPN303, slides 10–12)
+
+Status: numbers come from verified rehearsal takes (see `recording-runbook.md`). Re-check them against your final take; throughput varies a little from run to run. The script also serves as the shot list: each scene's **On screen** is what the recording must show.
+
+## Timing budget (from the deck's speaker notes)
+| Talk clock | Slide | Who | Length |
+|---|---|---|---|
+| 8:00 | 10: setup | Weiqing | 0:30 |
+| 8:30 | 11: demo video (muted) | Weiqing narrates | 7:00 |
+| 15:30 | slack (late start, a pause, a laugh) | | 0:30 |
+| 16:00 | 12: Agent ops patterns | Weiqing | 2:00 |
+
+Pace: about 120 words per minute. Each scene's lines are shorter than the scene, so the video never has to wait for you. Lines marked *(optional)* are the first to drop if you fall behind.
+
+Silence is planned. Scenes 3–6 are about half narration: let the audience watch the redeploy, the kill and the hold frames without talking over them.
+
+## How to stay in sync
+- Each scene opens with a 2-second chapter card ("1 · The runbook"). It's a resync point: if you're behind, jump to that scene's first line.
+- Each key moment gets a 2–3 second hold frame: bars even out, KEEP, ESCALATE, "redeploys: 1". Land the punchline on the hold.
+- On-screen highlight boxes do the pointing, so you never need a laser pointer.
+- Waits are time-lapsed, with a wall clock or "×N" badge on screen. Say "time-lapsed" once, in scene 3.
+
+---
+
+## Hand-in
+Mayank (end of slide 9): *"Weiqing, show them what a Flink job on call looks like."*
+
+## Slide 10: setup (0:30)
+**Proposed new slide text.** The current slide describes the old demo: vector store, four sample jobs, DashScope→Bedrock prep.
+- Title: Flink watching Flink
+- Subtitle: An operations agent, itself a Flink job, follows a stateful runbook for the jobs it watches.
+- 01 WAIT: Is it a real bottleneck, or a blip?
+- 02 EXPERIMENT: Baseline, change one setting, verify.
+- 03 REMEMBER: Every attempt lives in per-job Flink state.
+- 04 STOP: Keep, revert, or hand off with evidence.
+- Footer: EVERY RUN ENDS ONE WAY: KEPT · REVERTED · ESCALATED TO A PERSON
+
+**Say:**
+> Thanks, Mayank. This is a recording, so I'll narrate.
+> The agent is a Flink job, watching other Flink jobs. It isn't answering a question. It's following a runbook that takes several tries, so it has to remember what it did.
+> Watch for three things: it waits, it experiments, and it knows when to stop.
+
+*(Click: start the video.)*
+
+---
+
+## Scene 1: the runbook (0:00–0:45)
+**On screen:** the Flink 2.2 docs, Balanced Tasks Scheduling. Four phrases highlight one at a time, about 8s each.
+**Cue:** the first highlight appears.
+
+**Say:**
+> This is a real runbook: the Flink 2.2 docs for balanced task scheduling.
+> "If you're seeing bottlenecks": that's over time, not one snapshot.
+> "It may degrade": so take a baseline, and be ready to revert.
+> "Increase by 50 milliseconds each time": so remember what you tried.
+> "If it's still unbalanced, report it": with the history as evidence.
+> Every step depends on the one before. The runbook is stateful.
+
+## Scene 2: the agent is a Flink job (0:45–1:30)
+**On screen:** the architecture diagram (5s), then the Flink UI job graph: Kafka source → windows → persistence filter → agent operator at parallelism 4. Then the dashboard's header counters: metric samples → windows to the agent → LLM calls → side effects (a hold frame from the end of take A works best here).
+**Cue:** the job graph appears.
+
+**Say:**
+> Here's the agent: one Flink job. Metrics from every watched job stream in from Kafka, keyed by job name.
+> Windows and a filter run first, so the model only sees problems that persist. Look at the funnel: by the end of this run, about 3,500 metric samples, fewer than 30 windows reach the agent, and just 5 model calls.
+> Then the agent operator, at parallelism four. Each job's memory is Flink keyed state, checkpointed with the stream.
+
+## Scene 3: the page, and the wait (1:30–2:45)
+**On screen:** the target cluster's per-TaskManager CPU bars (one at 100% with 4 tasks, one with 2) and backpressure on ClickstreamEnrichment in the Flink UI. Then the agent's memory timeline, time-lapsed: window 1 skewed, window 2 skewed, window 3 skewed, baseline recorded.
+**Cue:** the red bar.
+
+**Say:**
+> Now the problem. ClickstreamEnrichment is backpressured. One TaskManager is pinned at 100 percent with four tasks; another has two.
+> An autoscaler would add machines. But this cluster isn't too small. The work is badly placed.
+> The agent doesn't react to one bad window. This part is time-lapsed. One window, two, three: the skew persists.
+> Only then does it act, and first it writes a baseline into memory: throughput, and tasks per TaskManager.
+
+## Scene 4: the fix, verified (2:45–4:00)
+**On screen:** the dashboard's *Latest decision* card (`APPLY_TASKS`, the runbook step, the rationale, "source: LLM"). Then the *Deployment API* card ticking through stop-with-savepoint → recreate cluster → submit from savepoint → RUNNING (time-lapsed; optional cutaway to `ctx.durable_execute_async` in `agent-job/ops_agent.py`), and the bars evening out to 3/3/3. Then the verify decision, KEEP: baseline ~7.5k records/s → 9.5k, the job's full input rate, with backpressure gone (optional c [...]
+**Cue:** the APPLY_TASKS decision appears.
+
+**Say:**
+> The decision comes back structured: which runbook step, which action, and why.
+> The action: switch the cluster to TASKS load balancing. That's a redeploy, a real side effect, so it runs as a durable action, with a stable request ID and its result checkpointed with the agent's state.
+> The cluster restarts from a savepoint, and... three, three, three.
+> The agent checks against its own baseline: balanced, backpressure gone, and the job now keeps up with its full input. Decision: keep.
+> Same three machines. No new hardware.
+
+*(Rehearsals measured a baseline of 6.4k–7.6k records/s → 9.5k, i.e. +25% to +48%. If your take shows a clear number, say it instead.)*
+
+## Scene 5: when the fix doesn't hold (4:00–5:15)
+**On screen:** the label **SIMULATED FLEET** stays visible the whole scene. The `payments-sessionizer` tile turns red: TASKS already on, skew back after 2 failovers. Memory timeline: interval 20 → 70 ms → still skewed → 120 → 170 → ESCALATE ("guardrail (code, no LLM)"). Then the escalation report: a drafted FLINK-38715 comment with an attempts table built from memory. **Hold** on ESCALATE.
+**Cue:** the SIMULATED label.
+
+**Say:**
+> Not every fix holds. This one is a simulated fleet job. TASKS is already on, but after two failovers the skew is back.
+> The runbook says: raise the slot request interval by 50 milliseconds and check again. The agent knows the current value and how many times it has tried, because that's in its memory.
+> Seventy. Still skewed. One-twenty. Still skewed. One-seventy.
+> Then it stops. That limit is code, not a prompt.
+> It escalates, and the report writes itself from memory: every attempt, every measurement, ready for a human.
+> *(optional)* And from then on it's hands-off: it keeps watching, but changes nothing until a person steps in.
+
+## Scene 6: break the agent (5:15–6:30)
+**On screen:** a terminal running `kill -9` on the agent's TaskManager while a redeploy is in flight. The Flink UI goes RESTARTING → RUNNING from the checkpoint (time-lapsed). The dashboard's Deployment API card shows the green note "agent recovered: its reconciler found this request and waited for it instead of sending it again", and **redeploys executed by the platform: 1**. The memory timeline is intact. **Hold** on "redeploys: 1".
+**Cue:** the kill command.
+
+**Say:**
+> Now let's break the agent. I kill its TaskManager in the middle of an action, while the redeploy is in flight.
+> Flink restarts the job from the last checkpoint. Two questions.
+> Does it redeploy the target twice? No. On recovery, the durable action checks the platform for its request ID, finds it already running, and waits for it instead of sending it again. One redeploy.
+> Does it forget? No. The baseline and every attempt come back with the checkpoint.
+> *(optional)* In a plain process you'd need your own database and an idempotency scheme for this. Here it's the engine.
+
+## Scene 7: close (6:30–7:00)
+**On screen:** a summary card: "Waited. Experimented. Remembered. Stopped." with outcome counts KEPT 1 · REVERTED 0 · ESCALATED 1.
+**Cue:** the summary card.
+
+**Say:**
+> So it didn't just answer. It waited for evidence, ran an experiment, remembered every step, survived a crash, and knew when to hand off to a person.
+> No external database, no scheduler, no queue. Just Flink state.
+
+*(Click: slide 12.)*
+
+---
+
+## Bridge into slide 12
+Replace the first line of slide 12's notes ("what you just saw is closest to pattern one") with:
+> That was one job and one runbook. You just saw all three patterns in miniature: the stream filtered before the model reasoned, each job's context lived in keyed state, and the guardrails were code. Here's how they look at LinkedIn scale.
+
+## Say it, don't show it (slide 12 or Q&A)
+- Scale: the agent is keyed by job name, so thousands of jobs spread across subtasks. Each key's actions stay in order, which is also what makes exactly-once actions possible after recovery.
+- Memory has a TTL, so a retired job's state expires on its own.
+- In production the "redeploy" is a Kubernetes operator spec change, gated by approvals and blast-radius limits.
+- The model is a local open model (Ollama qwen3:8b); any supported provider, Bedrock included, is a config change.
+
+## Q&A prep
+- **Why not the autoscaler?** It's a fixed formula on one metric and one setting, parallelism. For skew it adds machines when the problem is placement. The agent follows a written runbook across several settings, runs an experiment, and escalates with evidence.
+- **Why not a regular agent framework?** It's request-driven: you'd add a database for memory, a scheduler to wait across windows, a queue and partitioner for many jobs, and idempotency for side effects. Here those are keyed state, windows, key partitioning, and durable execution in one job.
+- **Is the fleet job real?** The hero job is real Flink on a real cluster. The fleet job is simulated, because the post-failover imbalance it shows doesn't reproduce on demand. The video labels it.
+- **Which version?** flink-agents built from main plus the durable-execution reconciler fix (apache/flink-agents#1161), heading into 0.4, on Flink 2.2.
+
+## Deck changes this script needs
+- Slide 10: replace the text (above) and the notes (drop the vector-store, DashScope→Bedrock, and manual-trigger prep).
+- Slide 11: embed the muted video, starting on click. Notes: "[8:30–15:30] narrate from demo-script.md". Keep a local copy of the video file on the presenting laptop.
+- Slide 12: the new tie-back line (above).
+- Slide 18 (hidden backup): no longer needed, since the video is the demo.
diff --git a/flink-stateful-ops-agent-demo/docs/plan.md b/flink-stateful-ops-agent-demo/docs/plan.md
new file mode 100644
index 0000000..92e5287
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/docs/plan.md
@@ -0,0 +1,226 @@
+# Plan: new demo module `flink-stateful-ops-agent-demo` (re:Invent OPN303, slides 10–12)
+
+## Decisions (2026-10-02)
+- Direction: build a new stateful tuning demo around the balanced-scheduling runbook. The original `flink-operations-agent-demo` stays untouched.
+- Demo format (2026-10-03): a **pre-recorded video** (7:00, slide 11), not a live demo. No venue network, latency or live-failure risk. Real waits (30s windows, redeploys, the ~65s meter lag) are time-lapsed in editing with an on-screen wall clock. Presenter narrates live over the muted video from `docs/demo-script.md`, so it can be rehearsed.
+- LLM (2026-10-03): record with **Ollama qwen3:8b**. It is verified 21/21 and needs no account, so anyone can reproduce the video from the public repo. Bedrock is an optional upgrade if Mayank (AWS) provides an account, and it doesn't block anything.
+- Target cluster: Docker, with a JobManager and 3 TaskManagers, each TaskManager limited to one CPU.
+- Current scope: P0 test, then the P1 skeleton, then check in.
+- flink-agents version: a local build of main + PR #1161, until a release contains the fix. Without it there's no "no double redeploy" beat (see P0 findings).
+- The hero job's live fix is NONE → TASKS on unbounded input. The "+50 ms" step stays on the simulated fleet job, because post-failover imbalance didn't reproduce on the hero job (see P0 findings: skew).
+
+## Demo story
+1. **The page.** ClickstreamEnrichment is backpressured. "An autoscaler would buy more machines."
+2. **The agent waits and remembers.**
+ - Per-job keyed state collects evidence across windows: one TaskManager is at 100% CPU with 4 tasks, while another has 2.
+ - It loads the runbook skill, records a baseline, and applies `TASKS` scheduling. That side effect runs exactly once.
+ - It verifies against the baseline: balanced, throughput up, so KEEP. No new machines.
+3. **When the fix doesn't hold** (simulated fleet job): +50 ms, then +50 ms, then it stops and escalates with a report built from its own memory.
+4. **Chaos.** Kill the agent mid-action. It recovers from the checkpoint, doesn't redeploy twice, and still remembers.
+5. **Close:** "It didn't just answer. It remembered, experimented, and knew when to stop." Then on to slide 12.
+
+## Q&A
+**Is the existing ops agent stateless?** Its reasoning is. It runs on a stateful engine but uses Flink only as a runner:
+- Only sensory memory, which is wiped after every inspection, so each check starts from zero.
+- Every job shares one key (`lambda x: 1`): no per-job memory, and inspections run one at a time.
+- Its only "history" is watermark JSON files written by an outside script, which aren't checkpointed.
+- No checkpoints or restart strategy: one LLM timeout killed the job.
+
+**How is this different from auto-sizing?**
+- An autoscaler is a fixed formula on one metric and one setting (parallelism), and it already exists in the Flink K8s Operator.
+- For skew it does the wrong thing: it adds resources, when the real problem is placement. The fix is the same resources, placed better.
+- The stateful agent follows a written runbook across several settings (scheduler mode, `slot.request.max-interval`, then a human).
+- It runs experiments: baseline → change → verify → keep or revert → escalate with evidence.
+- Each decision depends on what it has already tried. That is episodic memory, not a window of metrics.
+
+## Why a new demo (the original's weaknesses)
+- The original is basically auto-sizing. The Flink K8s Operator autoscaler already does that without an LLM, so expect the question "why not the autoscaler?"
+- It doesn't show anything stateful: one shared key, only sensory memory, history kept in files, no checkpoints or durability, and a cron-like trigger.
+
+## New demo case: "The runbook is stateful" (Flink 2.2 Balanced Tasks Scheduling, FLIP-370)
+Every line of the Flink doc's runbook depends on what happened before:
+1. "if you are seeing these bottlenecks": the bottleneck must persist across windows; a single snapshot isn't enough.
+2. "should not use it if not seeing bottlenecks, may degrade": compare against a baseline and revert if worse.
+3. "increase slot.request.max-interval by 50 ms each time": the agent needs the current value and the number of attempts.
+4. "if still unbalanced after sufficient increases, report in FLINK-38715": escalate with the full history as evidence.
+
+A stateless, request-driven agent needs an external DB, scheduler, queue and partitioner to follow this runbook. In Flink Agents it is per-job keyed state, checkpointed together with the stream.
+
+## Verified facts (2026-10-01)
+- Latest flink-agents release: 0.3.1 (PyPI). It ships dist jars for Flink 1.20, 2.0, 2.1, 2.2 and 2.3.
+- `taskmanager.load-balance.mode: TASKS` exists only in Flink ≥ 2.2. Flink 1.20 has only NONE/SLOTS.
+- `slot.request.max-interval` default is 20 ms (2.2).
+- In a Flink 2.2 session cluster, both options are read from the **cluster** config. `JobMasterServiceLeadershipRunnerFactory` passes the cluster `configuration` into `DefaultSlotPoolServiceSchedulerFactory.fromConfiguration`, so a job-level `-D` is ignored. The remedy is to redeploy the target cluster from a savepoint (in production, a FlinkDeployment spec change).
+- Stack: Flink 2.2.1, apache-flink 2.2.1, flink-connector-kafka 5.0.0-2.2, flink-agents 0.4-SNAPSHOT built locally from main + PR #1161 (0.4 is not released; see P0 findings: durability).
+- Features to use (all are in 0.3.1; the durability fixes are not):
+ - short-term memory plus TTL (`short-term-memory.state-ttl.ms`)
+ - action state store (Kafka) plus `durable_execute`
+ - Skills (`@skills`, `Skills.from_local_dir`, `load_skill`)
+ - event log (SLF4J, shown in the Flink UI Logs tab, or file)
+ - built-in metrics
+ - `output_schema`
+ - async/parallel LLM
+ - MCP
+ - Bedrock chat model (Java, used cross-language; optional, needs AWS credentials)
+
+## Architecture (a single Flink job; stream processing before reasoning)
+```
+collector (REST poll, 5s) ─┐
+fleet simulator (N jobs) ──┴─> Kafka flink_metrics
+ -> keyBy(job_name) -> 30s windows -> features (TM task skew, hot-TM busy%, backpressure, throughput)
+ -> KeyedProcessFunction: emit only if the anomaly persists ≥3 windows, or for follow-up windows after an incident
+ -> Flink Agents agent (keyed by job_name, parallelism 4)
+ memory: baseline, experiments[], current_config, phase, cooldown
+ skill: balanced-task-scheduling, loaded deterministically into every decision prompt
+ tools: redeploy_with_config (side effect, durable), get_tm_load
+ guardrails in code: cooldown, max attempts, then escalate
+ -> Kafka operations_record -> audit console (normal / auto-remediated / manual)
+```
+- Agent cluster: local Flink 2.2.1 standalone. It is never restarted by remediation.
+- Target cluster: docker compose with a JobManager and 3 TaskManagers (2 slots each, `cpus: 1`). The hero job has vertex A at p=6 and B at p=3, both CPU-heavy. The default strategy gives a 4/3/2 task skew, so the hot TaskManager backpressures.
+- Durability: checkpoint every 10s, fixed-delay restart, Kafka action state store.
+
+## Phases (each ends with something that can be demoed)
+
+**Status (2026-10-03):** P0–P4 done and verified end to end on the laptop (`TM_CPUS=0.5`): hero fix with a real redeploy and KEEP, fleet escalation with the hands-off hold, and the agent TaskManager kill mid-redeploy (restarts 1, redeploys 1, reconciled). The target job's input is rate-limited to `TARGET_RATE` (19000 × `TM_CPUS` = 9,500 events/s): the skewed placement falls behind (~6.5–7.5k rec/s, backpressured, hot SessionScore 100% busy), the balanced one keeps up (9.5k, no backpressu [...]
+
+- P0 spike, about 1 day. Verify on 2.2.1 + 0.3.1 Python:
+ - short-term memory persists across runs and checkpoints
+ - after a TaskManager kill, durable_execute does not replay the side effect
+ - Skills load
+ - event log shows in the UI
+ - structured output is reliable (Bedrock vs Ollama)
+ - the skew reproduces and TASKS mode fixes it
+- P1: new module skeleton on Flink 2.2 + main/#1161:
+ - key by job_name, parallelism 4
+ - checkpoints, restart strategy, Kafka action store
+ - memory: baseline, actions, config, cooldown, windows with skew
+ - the runbook loaded deterministically; `output_schema` decisions
+ - tools over `target_ctl`: redeploy (`durable_execute`), get_tm_load
+ - manual trigger, start/stop scripts, `setup_flink.sh` for the local build
+- P2: stream-first pipeline and fleet simulator (scale: N jobs, 4 subtasks, events→incidents→LLM-call counters).
+- P3: hero scenario:
+ - SOP as a Skill
+ - memory-driven loop: observe → baseline → apply TASKS → verify → keep/revert → +50 ms → escalate
+ - real target-cluster redeploy
+- P4: demo UX:
+ - per-job memory timeline view
+ - per-TaskManager CPU bars
+ - scene trigger scripts
+ - event log
+- P5: record and edit the demo video (the primary demo, slide 11). Shoot to `docs/demo-script.md`: its "On screen" lines are the shot list, with chapter cards and hold frames for live narration. Several takes per scene, time-lapse the waits, then a final cut. Hidden slide 18 is redundant.
+
+## Demo video flow (7:00, muted, narrated live; full script in `docs/demo-script.md`)
+Talk clock: slide 10 setup 8:00–8:30, video 8:30–15:30, 0:30 slack, slide 12 at 16:00.
+
+| Video time | Beat |
+|---|---|
+| 0:00 | The Flink doc runbook, highlighting the four stateful phrases |
+| 0:45 | Architecture plus Flink UI: agent operator at p=4; N jobs streaming; counters for events, incidents and LLM calls |
+| 1:30 | Hero job skewed (CPU bars). The agent waits 3 windows instead of reacting to a blip, and memory shows the baseline |
+| 2:45 | Agent applies TASKS (durable side effect), the target redeploys from a savepoint, and the bars even out. Verified against the baseline, so KEEP |
+| 4:00 | Fleet job (simulated) where the fix doesn't hold: +50 ms per try until the limit, then escalate with a report built from memory |
+| 5:15 | Kill the agent's TaskManager mid-action: it recovers from the checkpoint, does not redeploy twice, and memory is intact |
+| 6:30 | Summary card, then click to slide 12 |
+
+Say it rather than show it:
+- thousands of jobs through key partitioning
+- exactly-once action consistency needs the same per-key order after recovery
+- memory TTL
+- in production this is a K8s operator spec change, gated by approvals and blast-radius limits
+
+Recording rules:
+- Real runs only. Speed up waits but never fake a result, and label the simulated fleet job on screen.
+- Show a wall clock (or a "×N" badge) on every time-lapsed segment, so the 3-window wait stays honest.
+- Record at 1080p+ with large terminal and UI fonts; the video plays on a big screen.
+
+## Risks
+- ~~Whether the skew is deterministic on docker TaskManagers~~ Resolved: NONE skewed 5/5 times, TASKS balanced 4/4 (see P0 findings: skew).
+- Post-failover imbalance is not reproducible on demand: 0 of 3 TaskManager kills under TASKS. So the "+50 ms" chapter uses the simulated fleet and must be labeled honestly.
+- The Flink throughput meter lags about 65s behind a redeploy. Verify with `target_ctl.measure_throughput` (a 20s counter window) instead.
+- ~~Bedrock needs AWS credentials~~ Not blocking: the video is recorded with Ollama qwen3:8b (21/21, ~3.4s per decision). Bedrock is optional.
+- The recording and editing time must fit before the deck deadline; leave room for retakes.
+- **flink-agents version:** durability needs fixes that aren't in 0.3.1 (see P0 findings).
+
+## P0 findings (2026-10-02): memory and durable execution
+The spike lives in `spikes/`. The test kills the agent TaskManager with `kill -9` while actions are in flight.
+
+What we verified:
+- Short-term memory survives a TaskManager kill on every version: after restore, `runs` and `history` continue where they left off.
+- Per-key concurrency only holds with `async def` actions plus `await ctx.durable_execute_async`. A synchronous action blocks the subtask, so keys on that subtask run one after another.
+
+Durability results by version:
+
+| Version | Kill during APPLY, with reconciler | Kill during APPLY, no reconciler | Kill during OBSERVE (APPLY already done) |
+|---|---|---|---|
+| 0.3.1 | ❌ re-applies; restores 0 states | duplicate | ❌ re-applies |
+| main 99103da6 | ❌ only correct when no checkpoint lands mid-action | duplicate | ❌ same as left |
+| main + PR #1161 (14ee1feb) | ✅ the reconciler finds the receipt and skips | duplicate (expected baseline) | ✅ APPLY result replayed from the store; only OBSERVE re-runs |
+
+Upstream status:
+- **#1095 (merged):** cold-start topic-creation timeout.
+- **#1101 (merged):** the restore stopped at the first empty poll.
+- **#1158 (open):** the recovery marker is the end offset at snapshot time. Python async actions don't block the checkpoint barrier, so durable records written before a checkpoint by still-running actions get skipped. **PR #1161** fixes it and is verified end to end above: the restore now seeks to each in-flight key's latest record.
+- **#1174 (open):** the Fluss variant of #1158.
+
+**Decided:** pin the demo to main + #1161, then the first release that contains it. 0.3.1 can't show "no double redeploy".
+
+### Gotchas
+1. Don't copy flink-agents dist jars into `$FLINK_HOME/lib` for Python jobs. The duplicate kafka-clients cause a `JmxReporter` ClassCastException.
+2. Set `python.executable`, and export `PYTHONPATH=<venv purelib>` before starting TaskManagers (pemja).
+3. The agent class must live in an importable module. Submit with `-pym <main> -pyfs <absolute dir>`; a relative `-pyfs .` failed with ModuleNotFoundError.
+4. Add `pemja` to `classloader.parent-first-patterns.additional`. Otherwise every restart fails with "Native Library already loaded".
+5. The parent-first side effect: a TaskManager's Python module cache outlives the job. **Restart the cluster after changing agent code**, or the old code runs silently.
+6. The agent input must be Python objects. Put a `.map(lambda s: s)` after the Java KafkaSource.
+7. Read the input with `InputEvent.from_event(event).input`.
+8. Building flink-agents from source:
+ - compile on JDK 17 (output stays Java 11 bytecode)
+ - install the Python package with `FLINK_AGENTS_SKIP_JAR_DOWNLOAD=1`
+ - **use `mvn clean install` for the `dist/*` modules.** Without `clean`, maven-shade re-shades the previous jar and the stale runtime classes win.
+
+## P0 findings (2026-10-02): skew and remediation
+Setup:
+- Target cluster: `target-cluster/` (docker, 3 TaskManagers × 2 slots, `cpus: 1` each).
+- Hero job: `target-jobs/clickstream-enrichment` (Enrich p=6 → keyBy → SessionScore p=3, both CPU-heavy, unbounded datagen).
+- Control: `target-cluster/target_ctl.py`.
+- Trial runners: `spikes/skew_trials.py` and `spikes/failover_trials.py`.
+
+| Mode | Trials | Tasks per TaskManager | CPU per TaskManager | Throughput (SessionScore in) |
+|---|---|---|---|---|
+| NONE (default) | 5/5 skewed | 4/3/2 or 4/2/3 (the hot TaskManager varies) | 100% / ~68% / ~27% | ~17k rec/s; every Enrich subtask 70–90% backpressured |
+| TASKS | 4/4 balanced | 3/3/3 | 100% / 100% / 100% | ~27–28k rec/s (**+60% on the same 3 CPUs**) |
+| TASKS, after killing a TaskManager | 3/3 balanced | 3/3/3 | — | job recovered in 16–28s |
+
+- **Remediation:** redeploy the cluster from a savepoint, which takes about 13–16s (`target_ctl.redeploy`). Both options are cluster-level in a session cluster.
+- **Metrics:**
+ - CPU and busy-time metrics saturate about 16s after a redeploy.
+ - `numRecordsInPerSecond` is a 60s meter, so it lags about 65s.
+ - The JobManager's REST metric store answers from its *previous* fetch, so call `refresh_metrics()` before reading.
+ - `measure_throughput(jid, 20)` agrees with the meter to within 2%.
+- **Runbook source:**
+ - The Flink **2.2** docs only describe the mechanism, plus "don't use without bottlenecks".
+ - The failover note ("increase `slot.request.max-interval` by 50 ms each time … report in FLINK-38715") is in the **2.3/master** docs.
+ - The note comes from FLINK-38687 release testing. There the imbalance was random and needed several slot-sharing groups plus spare TaskManagers.
+ - The skill should cite both docs pages.
+- **Why not the autoscaler:** the backpressure is real, so an autoscaler would add capacity. But the fix is placement, on the same machines.
+
+## P0 findings (2026-10-02): LLM decisions
+Spike: `spikes/llm_spike.py` covers 7 runbook scenarios (WAIT, APPLY_TASKS, KEEP, REVERT, INCREASE_INTERVAL, ESCALATE, NO_ACTION). It runs through a flink-agents agent on a PyFlink mini-cluster with Ollama qwen3:8b at temperature 0.
+
+| Variant | Correct | Latency per call |
+|---|---|---|
+| **Runbook in the system prompt, `output_schema`, no thinking** | **21/21** (3 reps), 0 parse errors, 0 retries | median 3.4s, max 7.1s |
+| Same, thinking on | 7/7 | 13–60s (no accuracy gain; slows every take) |
+| Runbook as a Skill (model must call `load_skill`), with or without `output_schema`, with or without thinking | 2–3/7: never called `load_skill` | 2–40s |
+
+- The runbook needs an explicit **decision table** (situation → action) plus a `runbook_step` field placed before `action` in the output. With prose alone, the model jumps to APPLY_TASKS (4/7).
+- Ollama native structured output (`format`) constrains decoding to the JSON schema, so the model **cannot emit a tool call** while it's on. Even without it, qwen3:8b skips `load_skill` whenever the prompt also asks for JSON (0/6 in direct replays).
+- **Decision:** the runbook is still a standard Agent Skill (`skills/balanced-task-scheduling/SKILL.md`, citing the Flink docs), but the agent **loads it deterministically** into every decision prompt. The talking point: "an ops agent shouldn't get to choose whether to read the runbook." Re-test model-driven `load_skill` on Bedrock once credentials exist.
+- Python agent jobs: the agent class must be imported from a module, not `__main__`. For local runs use `python -c 'import mod; mod.main()'`, and set `python.executable` to the venv interpreter.
+- Event log: `baseLogDir` → `FileEventLogger` writes per-subtask JSONL. Strings are truncated at 2000 chars (`event-log.standard.max-string-length`).
+
+## Local dev safety (the presenter's MacBook: M3 Pro, 12 cores, 36 GB)
+- Develop with `TM_CPUS=0.5`, so the target cluster is hard-capped at 1.5 cores (~12% of the machine). Use the full `TM_CPUS=1` only for recording takes.
+- Ollama runs only during decisions (~3s bursts).
+- Run `target_ctl.py down`, stop the agent cluster and stop Kafka whenever a test session ends.
+- Check `pmset -g therm` for thermal warnings during long runs.
diff --git a/flink-stateful-ops-agent-demo/docs/recording-runbook.md b/flink-stateful-ops-agent-demo/docs/recording-runbook.md
new file mode 100644
index 0000000..cbf6dcd
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/docs/recording-runbook.md
@@ -0,0 +1,143 @@
+# Recording runbook
+
+How to record the muted demo video that `demo-script.md` narrates. Every command and
+timing below comes from verified rehearsal runs on an M3 Pro laptop with `TM_CPUS=0.5`
+(the default). Run all commands from `flink-agents-demos/flink-stateful-ops-agent-demo`.
+
+## What you record
+| Footage | Covers script scenes | Stack needed | Real length |
+|---|---|---|---|
+| Take A: hero fix, then fleet escalation | 2, 3, 4, 5 | yes | ~7 min |
+| Take B: kill the agent mid-redeploy | 6 | yes | ~3 min |
+| Screen shots | 1 (docs page), 2 (diagram), 7 (summary card) | no | ~1 min |
+
+Edit these into the 7:00 video: time-lapse the waits, add the chapter cards and hold frames.
+
+## 0. Before the session (once)
+Start **Docker Desktop** and the **Ollama** app first.
+```bash
+bin/demo.sh setup # only on a new machine: Flink, venv, flink-agents main + #1161, jar, model
+bin/demo.sh up # Kafka, target cluster (skewed), agent cluster, Ollama warm, platform-sim
+bin/demo.sh status # everything "up", target "skewed"
+```
+`up` takes about 1–2 minutes. The laptop load stays moderate: the three target TaskManagers
+are capped at 0.5 CPU each.
+
+## 1. Screen setup
+Use a 1920×1080 recording area: macOS **Cmd+Shift+5** → *Record Selected Portion*, or set the
+display to 1920×1080. Turn on Do Not Disturb. Quit Slack and mail. Hide the Chrome bookmarks bar.
+
+Open these before you start recording:
+
+| Where | What | Used in |
+|---|---|---|
+| Chrome tab 1 | Flink docs: https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/deployment/tasks-scheduling/balanced_tasks_scheduling/ | scene 1 |
+| Chrome tab 2 | **Dashboard** http://localhost:8090 (keys `1` = real job, `2` = fleet) | scenes 3–6 |
+| Chrome tab 3 | **Agent Flink UI** http://localhost:8082 → Running Jobs → `stateful-ops-agent` | scenes 2, 6 |
+| Chrome tab 4 | **Target Flink UI** http://localhost:8081 → `ClickstreamEnrichment` | scenes 3, 4 |
+| VS Code (optional cutaways) | `skills/balanced-task-scheduling/SKILL.md` (decision table, line 57) · `agent-job/ops_agent.py` (`durable_execute_async`, line 245) · `agent-job/ops_runbook.py` (`allowed_actions`, line 106) | scenes 1, 4, 5 |
+| iTerm, font ≥ 20pt | the commands below | all takes |
+
+The dashboard fits 1080p exactly. Chrome zoom 100%. Use full screen (Cmd+Ctrl+F) on the dashboard tab.
+
+**Target UI (8081), before vs after.** The click source emits a fixed 9,500 events/s
+(`TARGET_RATE`, default 19000 × `TM_CPUS`). The skewed placement can't keep up; the balanced one can.
+- Before (NONE, 4/3/2): `SessionScore -> Sink` is red (the hot subtask is 100% busy), `Source: Clicks -> Enrich`
+ is backpressured (300–600 ms/s), and the job does ~6.5–7.5k records/s.
+- After (TASKS, 3/3/3): backpressure 0 and 9.5k records/s. SessionScore busy falls to ~50% about 1 min after
+ the redeploy (JIT warm-up), then to ~25–40%. Take the "after" shot at or after KEEP.
+
+## 2. Take A: scenes 2–5
+**Reset, then start recording.**
+```bash
+bin/demo.sh reset # ~1.5–2 min. Ends with "Ready for a take".
+```
+`reset` cancels the agent, clears the dashboard, recreates the target cluster until NONE mode
+places tasks unevenly (it retries automatically), restarts the agent cluster, warms the model, and
+waits until the target falls behind its input (Enrich backpressure ≥ 300 ms/s).
+
+T = seconds after you run `bin/demo.sh agent`. Measured times; allow ±10 s.
+
+| T | Do / run | Look at | What it shows (script scene) |
+|---|---|---|---|
+| 0 | `bin/demo.sh agent` | terminal | "Job has been submitted" |
+| 5–15 | — | **Agent UI** (8082) → `stateful-ops-agent` → job graph | Kafka source → `window 15s` → `persistence filter` → `OpsAgent` at parallelism 4 (scene 2) |
+| 15–25 | — | **Target UI** (8081) → ClickstreamEnrichment job graph; click `Source: Clicks -> Enrich` → *BackPressure* tab | SessionScore red, Enrich backpressured (scene 3) |
+| 25 | press `1` | **Dashboard** | red bar: the TaskManager with 4 tasks at ~100% CPU, another with 2 (scene 3) |
+| ~18, ~34 | — | Dashboard timeline | `skewed 1/3 windows: WAIT`, `skewed 2/3 windows: WAIT` (scene 3, "it waits") |
+| ~52 | — | Dashboard: *Latest decision* flashes, *Agent memory* | `APPLY_TASKS`, `source: LLM (qwen3:8b, ~5–7 s)`; memory shows incident #1 and the **baseline** (~7.5k rec/s) (scene 4) |
+| 52–77 | — | Dashboard: *Deployment API* card ticks `stop-with-savepoint → recreate cluster → submit from savepoint → wait for RUNNING`; TM bars show "cluster restarting…" | the real redeploy, ~25 s (scene 4). Optional: Target UI shows the job coming back |
+| ~77 | — | Dashboard TM bars | **3 / 3 / 3**, all green. "redeploys executed by the platform: 1" |
+| 77–127 | — | timeline countdown `cooldown: verify in …s`, records/s rising to 9.5k | the agent waits out the cooldown before judging |
+| ~127 | **hold 3 s** | *Latest decision* | `KEEP`, throughput ≥ baseline (~7.5k → 9.5k rec/s, the full input rate); memory: `closed: #1 KEPT` (scene 4) |
+| 130–140 | — | **Target UI** (8081) job graph | after shot: no backpressure, SessionScore no longer red (scene 4, optional) |
+| ~140 | `bin/demo.sh fleet-incident`, then press `2` | **Dashboard tab 2** | SIMULATED FLEET label; `payments-sessionizer` tile turns red, TASKS already on, 2 failovers (scene 5) |
+
+Fleet timings, F = seconds after `fleet-incident`:
+
+| F | Look at | What it shows |
+|---|---|---|
+| ~25, ~40 | timeline | `skewed 1/3`, `skewed 2/3`: WAIT |
+| ~60 | *Latest decision*, memory | `INCREASE_INTERVAL` (LLM) → chip `slot.request.max-interval: 70ms`; memory: `increases 1 of 3` |
+| ~120 | same | `INCREASE_INTERVAL` → 120 ms; attempts table grows, each row with "after" measurements |
+| ~180 | same | `INCREASE_INTERVAL` → 170 ms |
+| ~236 | **hold 3 s** on *Escalation filed* + *Latest decision* | `ESCALATE` from `guardrail (code, no LLM)`; the FLINK-38715 report with the attempts table, written from memory |
+| ~250+ | memory note, timeline | `escalated as …: hands off` — the agent keeps watching but changes nothing |
+
+Good cutaways during the cooldowns (30–45 s of nothing new each): the SKILL.md decision table, and
+`allowed_actions` in `ops_runbook.py` ("the limit is code, not a prompt").
+
+**Stop recording.** Header counters at the end of a take: ~3,500 metric samples → 26–29 windows
+→ 5 LLM calls → 5 side effects; KEPT 1 · REVERTED 0 · ESCALATED 1.
+
+## 3. Take B: scene 6 (kill the agent)
+Put the terminal and the **Agent UI** job page side by side, with the dashboard (tab 1) one swipe away.
+```bash
+bin/demo.sh reset
+# start recording
+bin/demo.sh agent
+bin/demo.sh kill-agent-tm --wait
+```
+`--wait` blocks until the agent starts the redeploy (T ≈ 50 s). Five seconds later it prints and
+runs, for example:
+```
+Redeploy in flight: ClickstreamEnrichment#1.1 (ClickstreamEnrichment), sent by the agent TaskManager pid 41383
+$ kill -9 41383
+```
+
+| After the kill | Look at | What it shows |
+|---|---|---|
+| 0–15 s | Agent UI | job RESTARTING → RUNNING on the surviving TaskManager; dashboard header `restarts 1` |
+| ~25–30 s | Dashboard *Deployment API* card | steps finish (rehearsal: 25.0 s, marked `reconciled`); green note: *agent recovered: its reconciler found this request and waited for it instead of sending it again* |
+| — | Dashboard | **redeploys executed by the platform: 1** (hold here). The platform does not de-duplicate, so a resend would show 2 |
+| — | Agent memory | baseline and attempt survived; the attempt is marked `reconciled` |
+| ~T+127 | Latest decision | KEEP, as in take A (rehearsal: T+127, baseline 7.6k → 9.5k rec/s) |
+
+Header counters at the end of take B: `restarts 1`, KEPT 1 · REVERTED 0 · ESCALATED 0.
+
+## 4. Screen-only shots
+- Scene 1: Chrome tab 1. Select each quoted phrase with the mouse as you scroll; the selection is the highlight.
+- Scene 2: the architecture diagram from the deck (5 s).
+- Scene 7: a title card in the editor: "Waited. Experimented. Remembered. Stopped." KEPT 1 · REVERTED 0 · ESCALATED 1.
+
+## 5. Editing to 7:00
+Map: scene 2 ← take A T 0–25 · scene 3 ← T 25–52 · scene 4 ← T 52–140 · scene 5 ← fleet F 0–250 ·
+scene 6 ← take B. Time-lapse the countdowns ×4–×8 with a "×N" badge, add 2 s chapter cards and
+2–3 s holds on: bars at 3/3/3, KEEP, ESCALATE, "redeploys: 1". iMovie or ScreenFlow is enough.
+
+## Retakes and troubleshooting
+- Anything off: `bin/demo.sh reset` and go again. Each take starts clean.
+- Between takes use `reset`, not `up`. `up` re-skews the target but leaves an old agent job running, which would start acting on it.
+- `Target placement did not skew after 3 attempts`: run `bin/demo.sh reset` again.
+- `An agent job is already running`: `bin/demo.sh reset`.
+- Dashboard says agent `NO JOB`: `bin/demo.sh status`; logs in `tmp/platform/platform.log` and `flink-2.2.1/log/`.
+- First LLM call slow (> 20 s): the model was unloaded; `reset` warms it, or run `bin/demo.sh up`.
+- `reset` warns `target not backpressured yet`: the skew windows start later than the table says; wait ~30 s before `agent`, or `reset` again.
+- Target still red after the fix (or never backpressured before it): your CPU differs from the rehearsal machine. Set `TARGET_RATE` (lower if red after, higher if not backpressured before) and `reset`.
+- After take B the agent cluster has one TaskManager left; `reset` restores two.
+
+## Shut down
+```bash
+bin/demo.sh down # clusters + platform-sim (Kafka and the model stay)
+bin/demo.sh down --all # also Kafka, and unload the model
+```
diff --git a/flink-stateful-ops-agent-demo/platform-sim/collector.py b/flink-stateful-ops-agent-demo/platform-sim/collector.py
new file mode 100644
index 0000000..b3a8d59
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/platform-sim/collector.py
@@ -0,0 +1,86 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""One metric sample of the REAL target job, read from the target cluster's REST API.
+
+The REST metric store answers from its previous fetch, so values are a few seconds
+old; that is fine for windows of 15s.
+"""
+import json
+import urllib.request
+
+import target_ctl
+
+TIMEOUT_S = 2
+
+
+def _get(path):
+ with urllib.request.urlopen(target_ctl.REST + path, timeout=TIMEOUT_S) as resp:
+ return json.loads(resp.read() or b"{}")
+
+
+def _agg(jid, vid, metric, agg):
+ values = _get(f"/jobs/{jid}/vertices/{vid}/subtasks/metrics?get={metric}&agg={agg}")
+ return float(values[0][agg]) if values else None
+
+
+def _parse_ms(value):
+ value = (value or "20ms").strip()
+ if value.endswith("ms"):
+ return int(float(value[:-2]))
+ if value.endswith("s"):
+ return int(float(value[:-1]) * 1000)
+ return int(float(value))
+
+
+def sample(job_key, ts):
+ s = {"job": job_key, "simulated": False, "ts": ts, "reachable": False, "running": False}
+ try:
+ cfg = {e["key"]: e["value"] for e in _get("/jobmanager/config")}
+ s["reachable"] = True
+ s["config"] = {"mode": cfg.get("taskmanager.load-balance.mode", "NONE"),
+ "interval_ms": _parse_ms(cfg.get("slot.request.max-interval"))}
+ jid = next((j["jid"] for j in _get("/jobs/overview")["jobs"]
+ if j["name"] == target_ctl.JOB_NAME and j["state"] == "RUNNING"), None)
+ if not jid:
+ return s
+ s["job_id"] = jid
+ vertices = _get(f"/jobs/{jid}")["vertices"]
+ if not all(v["status"] == "RUNNING" for v in vertices):
+ return s
+ tms = sorted(tm["id"] for tm in _get("/taskmanagers")["taskmanagers"])
+ tasks = {tm: 0 for tm in tms}
+ for v in vertices:
+ for st in _get(f"/jobs/{jid}/vertices/{v['id']}")["subtasks"]:
+ if st["taskmanager-id"] not in tasks:
+ return s
+ tasks[st["taskmanager-id"]] += 1
+ cpu = {}
+ for tm in tms:
+ values = _get(f"/taskmanagers/{tm}/metrics?get=Status.JVM.CPU.Load")
+ if not values:
+ return s
+ cpu[tm] = round(float(values[0]["value"]), 3)
+ backpressure = _agg(jid, vertices[0]["id"], "backPressuredTimeMsPerSecond", "max")
+ records_in = _agg(jid, vertices[-1]["id"], "numRecordsIn", "sum")
+ if backpressure is None or records_in is None:
+ return s
+ s.update(running=True, tasks_per_tm=tasks, cpu=cpu, backpressure_ms=round(backpressure),
+ records_in=records_in, recent_failovers=0)
+ except (OSError, ValueError, KeyError):
+ pass
+ return s
diff --git a/flink-stateful-ops-agent-demo/platform-sim/dashboard.html b/flink-stateful-ops-agent-demo/platform-sim/dashboard.html
new file mode 100644
index 0000000..59cd78c
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/platform-sim/dashboard.html
@@ -0,0 +1,350 @@
+<!doctype html>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<html lang="en">
+<head>
+<meta charset="utf-8">
+<title>Flink Agents · operations agent</title>
+<style>
+ :root {
+ --bg: #0d1117; --card: #161b22; --line: #30363d; --text: #e6edf3; --dim: #8b949e;
+ --red: #f85149; --amber: #d29922; --green: #3fb950; --blue: #58a6ff; --purple: #bc8cff;
+ }
+ * { box-sizing: border-box; }
+ body { margin: 0; background: var(--bg); color: var(--text); font: 18px/1.35 -apple-system, "SF Pro Text", "Segoe UI", sans-serif;
+ height: 100vh; overflow: hidden; display: flex; flex-direction: column; }
+ header { display: grid; grid-template-columns: auto 1fr auto; gap: 24px; align-items: center; padding: 12px 24px; border-bottom: 1px solid var(--line); }
+ h1 { font-size: 24px; margin: 0; }
+ h1 small { display: block; color: var(--dim); font-size: 15px; font-weight: 400; }
+ .funnel { display: flex; justify-content: center; align-items: center; gap: 10px; }
+ .count { background: var(--card); border: 1px solid var(--line); border-radius: 10px; padding: 6px 16px; text-align: center; min-width: 150px; }
+ .count b { display: block; font-size: 30px; font-variant-numeric: tabular-nums; }
+ .count span { color: var(--dim); font-size: 14px; }
+ .arrow { color: var(--dim); font-size: 26px; }
+ .clock { font-size: 34px; font-variant-numeric: tabular-nums; font-weight: 600; text-align: right; }
+ .clock small { display: block; font-size: 14px; font-weight: 400; color: var(--dim); }
+ nav { display: flex; gap: 8px; padding: 10px 24px 0; align-items: flex-end; }
+ .outcomes { margin-left: auto; display: flex; gap: 28px; font-size: 18px; color: var(--dim); padding-bottom: 6px; }
+ .outcomes b { color: var(--text); font-size: 24px; margin-left: 6px; }
+ nav button { background: none; border: 1px solid var(--line); color: var(--dim); border-radius: 8px 8px 0 0; padding: 8px 18px; font-size: 17px; cursor: pointer; }
+ nav button.on { color: var(--text); background: var(--card); border-bottom-color: var(--card); }
+ main { display: grid; grid-template-columns: 1fr 1fr; gap: 16px; padding: 0 24px 12px; flex: 1; min-height: 0; }
+ main.hidden { display: none; }
+ .col { display: flex; flex-direction: column; gap: 14px; min-width: 0; min-height: 0; overflow: hidden; }
+ .card { flex-shrink: 0; }
+ .card { background: var(--card); border: 1px solid var(--line); border-radius: 12px; padding: 14px 18px; }
+ .card h2 { font-size: 15px; text-transform: uppercase; letter-spacing: .06em; color: var(--dim); margin: 0 0 10px; font-weight: 600; }
+ .title { font-size: 26px; font-weight: 700; display: flex; align-items: center; gap: 12px; flex-wrap: wrap; }
+ .badge { font-size: 14px; font-weight: 700; padding: 3px 10px; border-radius: 6px; letter-spacing: .04em; }
+ .real { background: rgba(63,185,80,.15); color: var(--green); border: 1px solid var(--green); }
+ .sim { background: rgba(210,153,34,.18); color: var(--amber); border: 2px solid var(--amber); font-size: 18px; }
+ .chips { display: flex; gap: 8px; flex-wrap: wrap; margin: 10px 0 4px; }
+ .chip { font-family: ui-monospace, Menlo, monospace; font-size: 15px; background: #0d1117; border: 1px solid var(--line); border-radius: 6px; padding: 3px 8px; }
+ .chip.hl { border-color: var(--blue); color: var(--blue); }
+ .tm { display: grid; grid-template-columns: 70px 170px 1fr 64px; gap: 12px; align-items: center; margin: 10px 0; }
+ .tm .name { font-weight: 600; }
+ .tasks { display: flex; gap: 4px; align-items: center; }
+ .sq { width: 20px; height: 20px; border-radius: 4px; background: var(--blue); }
+ .tasks em { color: var(--dim); font-style: normal; font-size: 15px; margin-left: 6px; }
+ .bar { height: 22px; background: #0d1117; border-radius: 6px; overflow: hidden; border: 1px solid var(--line); }
+ .bar div { height: 100%; transition: width .8s, background .8s; }
+ .pct { text-align: right; font-variant-numeric: tabular-nums; font-weight: 600; }
+ .stats { display: grid; grid-template-columns: repeat(3, 1fr); gap: 10px; margin-top: 12px; }
+ .stat { background: #0d1117; border: 1px solid var(--line); border-radius: 10px; padding: 8px 12px; }
+ .stat b { display: block; font-size: 28px; font-variant-numeric: tabular-nums; }
+ .stat span { color: var(--dim); font-size: 14px; }
+ .big { font-size: 48px !important; line-height: 1; }
+ .steps { list-style: none; padding: 0; margin: 8px 0 0; display: flex; flex-wrap: wrap; gap: 2px 18px; }
+ .steps li { padding: 2px 0; }
+ .steps li.done::before { content: "✓ "; color: var(--green); }
+ .steps li.run::before { content: "◌ "; color: var(--amber); }
+ .note { margin-top: 8px; padding: 8px 10px; border-radius: 8px; font-size: 16px; }
+ .note.ok { background: rgba(63,185,80,.12); border: 1px solid var(--green); }
+ .note.bad { background: rgba(248,81,73,.12); border: 1px solid var(--red); }
+ .note.hold { background: rgba(210,153,34,.14); border: 1px solid var(--amber); }
+ table { border-collapse: collapse; width: 100%; font-size: 16px; }
+ td, th { text-align: left; padding: 4px 8px 4px 0; border-bottom: 1px solid var(--line); vertical-align: top; }
+ th { color: var(--dim); font-weight: 500; }
+ .mono { font-family: ui-monospace, Menlo, monospace; font-size: 15px; }
+ .kv { display: grid; grid-template-columns: 150px 1fr; gap: 4px 10px; }
+ .kv span { color: var(--dim); }
+ pre { margin: 0; white-space: pre-wrap; font-size: 16px; font-family: ui-monospace, Menlo, monospace; }
+ .flash { animation: flash 2.5s ease-out; }
+ @keyframes flash { from { box-shadow: 0 0 0 3px var(--blue); } to { box-shadow: 0 0 0 0 transparent; } }
+ .tl { display: grid; grid-template-columns: 90px 96px 1fr; gap: 8px; padding: 4px 0; border-bottom: 1px solid var(--line); font-size: 16px; }
+ .kind { font-size: 13px; font-weight: 700; text-transform: uppercase; border-radius: 5px; padding: 1px 6px; text-align: center; align-self: start; }
+ .k-observe { color: var(--dim); border: 1px solid var(--line); }
+ .k-decision { color: var(--blue); border: 1px solid var(--blue); }
+ .k-applied { color: var(--green); border: 1px solid var(--green); }
+ .k-escalated { color: var(--amber); border: 1px solid var(--amber); }
+ .dim { color: var(--dim); }
+ .tiles { display: grid; grid-template-columns: repeat(7, 1fr); gap: 6px; }
+ .tile { border-radius: 6px; padding: 5px 6px; font-size: 12px; line-height: 1.3; height: 40px; overflow: hidden; white-space: nowrap; text-overflow: ellipsis; border: 1px solid var(--line); background: rgba(63,185,80,.10); }
+ .tile.unbalanced { background: rgba(248,81,73,.25); border-color: var(--red); }
+ .tile.redeploying { background: rgba(210,153,34,.25); border-color: var(--amber); }
+ .tile.focus { outline: 2px solid var(--text); }
+ .report h4 { margin: 0 0 6px; font-size: 18px; }
+ .report ul { margin: 4px 0 8px; padding-left: 20px; }
+ .report code { font-size: 14px; }
+ .agent-state { font-weight: 700; }
+ .s-RUNNING { color: var(--green); } .s-RESTARTING, .s-INITIALIZING, .s-CREATED { color: var(--amber); }
+ .s-FAILED, .s-CANCELED, .s-UNREACHABLE { color: var(--red); }
+ .hidden { display: none; }
+</style>
+</head>
+<body>
+<header>
+ <h1>Flink Agents · operations agent <small>a Flink job on call for Flink jobs · agent job <span id="agent" class="agent-state">–</span></small></h1>
+ <div class="funnel">
+ <div class="count"><b id="c-samples">0</b><span>metric samples</span></div><div class="arrow">→</div>
+ <div class="count"><b id="c-windows">0</b><span>windows to the agent</span></div><div class="arrow">→</div>
+ <div class="count"><b id="c-llm">0</b><span>LLM calls</span></div><div class="arrow">→</div>
+ <div class="count"><b id="c-effects">0</b><span>side effects</span></div>
+ </div>
+ <div class="clock"><span id="clock">--:--:--</span><small>wall clock</small></div>
+</header>
+<nav>
+ <button id="tab-hero" class="on" onclick="show('hero')">1 · ClickstreamEnrichment (real)</button>
+ <button id="tab-fleet" onclick="show('fleet')">2 · fleet (simulated)</button>
+ <div class="outcomes"><span>KEPT<b id="o-KEPT">0</b></span><span>REVERTED<b id="o-REVERTED">0</b></span><span>ESCALATED<b id="o-ESCALATED">0</b></span></div>
+</nav>
+
+<main id="view-hero">
+ <div class="col">
+ <div class="card">
+ <div class="title">ClickstreamEnrichment <span class="badge real">REAL · Flink 2.2 cluster</span></div>
+ <div class="chips" id="h-config"></div>
+ <div id="h-tms"></div>
+ <div class="stats">
+ <div class="stat"><b id="h-rate">–</b><span>records/s (live)</span></div>
+ <div class="stat"><b id="h-bp">–</b><span>Enrich backpressured ms/s</span></div>
+ <div class="stat"><b id="h-redeploys" class="big">0</b><span>redeploys executed by the platform</span></div>
+ </div>
+ </div>
+ <div class="card" id="h-deploy-card"><h2>Deployment API (no de-duplication)</h2><div id="h-deploy" class="dim">no request yet</div></div>
+ </div>
+ <div class="col">
+ <div class="card" id="h-memory-card"><h2>Agent memory · Flink keyed state for this job</h2><div id="h-memory" class="dim">empty</div></div>
+ <div class="card" id="h-decision-card"><h2>Latest decision</h2><div id="h-decision" class="dim">none yet</div></div>
+ <div class="card"><h2>Timeline · what the agent did, window by window</h2><div id="h-timeline" class="dim">waiting for the first window</div></div>
+ </div>
+</main>
+
+<main id="view-fleet" class="hidden">
+ <div class="col">
+ <div class="card">
+ <div class="title"><span class="badge sim">SIMULATED FLEET</span> <span id="f-count"></span></div>
+ <div class="tiles" id="f-tiles" style="margin-top:12px"></div>
+ </div>
+ <div class="card">
+ <div class="title" id="f-name"></div>
+ <div class="chips" id="f-config"></div>
+ <div id="f-tms"></div>
+ <div class="stats">
+ <div class="stat"><b id="f-rate">–</b><span>records/s (window)</span></div>
+ <div class="stat"><b id="f-failovers">–</b><span>recent failovers</span></div>
+ <div class="stat"><b id="f-redeploys" class="big">0</b><span>redeploys executed</span></div>
+ </div>
+ <div id="f-deploy"></div>
+ </div>
+ <div class="card" id="f-decision-card"><h2>Latest decision</h2><div id="f-decision" class="dim">none yet</div></div>
+ </div>
+ <div class="col">
+ <div class="card" id="f-memory-card"><h2>Agent memory · Flink keyed state for this job</h2><div id="f-memory" class="dim">empty</div></div>
+ <div class="card" id="f-escalation-card"><h2>Escalation filed</h2><div id="f-escalation" class="dim report">none</div></div>
+ <div class="card"><h2>Timeline</h2><div id="f-timeline" class="dim">waiting</div></div>
+ </div>
+</main>
+
+
+<script>
+const $ = (id) => document.getElementById(id);
+const esc = (s) => String(s ?? "").replace(/[&<>"]/g, (c) => ({ "&": "&", "<": "<", ">": ">", '"': """ }[c]));
+const num = (n) => (n === null || n === undefined) ? "–" : Math.round(n).toLocaleString("en-US");
+const hms = (t) => t ? new Date(t * 1000).toLocaleTimeString("en-GB") : "";
+const cpuColor = (c) => c >= 0.85 ? "var(--red)" : c >= 0.6 ? "var(--amber)" : "var(--green)";
+const seen = {};
+
+function show(view) {
+ for (const v of ["hero", "fleet"]) {
+ $("view-" + v).classList.toggle("hidden", v !== view);
+ $("tab-" + v).classList.toggle("on", v === view);
+ }
+ location.hash = view;
+}
+document.addEventListener("keydown", (e) => { if (e.key === "1") show("hero"); if (e.key === "2") show("fleet"); });
+if (location.hash === "#fleet") show("fleet");
+
+function setHTML(id, html) { if ($(id).innerHTML !== html) $(id).innerHTML = html; }
+function flashOnChange(cardId, key) {
+ if (key && seen[cardId] !== key) {
+ if (seen[cardId] !== undefined) { const el = $(cardId); el.classList.remove("flash"); void el.offsetWidth; el.classList.add("flash"); }
+ seen[cardId] = key;
+ }
+}
+
+function configChips(config, extra) {
+ if (!config) return '<span class="chip">config unknown</span>';
+ return `<span class="chip hl">taskmanager.load-balance.mode: ${esc(config.mode)}</span>` +
+ `<span class="chip">slot.request.max-interval: ${esc(config.interval_ms)}ms</span>` + (extra || "");
+}
+
+function tmRows(tasks, cpu) {
+ if (!tasks) return '<div class="dim">job not running</div>';
+ const counts = Object.values(tasks), balanced = Math.max(...counts) - Math.min(...counts) < 2;
+ return Object.keys(tasks).sort().map((tm) => {
+ const c = cpu ? cpu[tm] : null, n = tasks[tm];
+ return `<div class="tm"><div class="name">${esc(tm)}</div>
+ <div class="tasks">${'<div class="sq"></div>'.repeat(n)}<em>${n} tasks</em></div>
+ <div class="bar"><div style="width:${Math.round((c || 0) * 100)}%;background:${balanced ? "var(--green)" : cpuColor(c || 0)}"></div></div>
+ <div class="pct">${c === null || c === undefined ? "–" : Math.round(c * 100) + "%"}</div></div>`;
+ }).join("");
+}
+
+function deployHTML(d) {
+ if (!d) return '<span class="dim">no request yet</span>';
+ const steps = d.steps.map((s) => `<li class="${s.done ? "done" : "run"}">${esc(s.name)}</li>`).join("");
+ const notes = (d.notes || []).map((n) => `<div class="note ${n.startsWith("DUPLICATE") ? "bad" : "ok"}">${esc(n)}</div>`).join("");
+ return `<div class="kv"><span>request</span><b class="mono">${esc(d.request_id)}</b>
+ <span>target</span><span class="mono" style="color:var(--text)">${esc(d.target.mode)}, ${esc(d.target.interval_ms)}ms</span>
+ <span>status</span><b>${esc(d.status)}${d.took_s ? ` in ${d.took_s}s` : ""}</b></div>
+ <ul class="steps">${steps}</ul>${notes}`;
+}
+
+function decisionHTML(d) {
+ if (!d) return '<span class="dim">none yet</span>';
+ const src = d.source === "llm" ? `LLM (qwen3:8b, ${d.llm_seconds}s)` : (d.llm_seconds ? `guardrail after the LLM (${d.llm_seconds}s)` : "guardrail (code, no LLM)");
+ return `<pre>${esc(JSON.stringify(d.decision, null, 2))}</pre>
+ <div class="dim" style="margin-top:6px">${hms(d.at)} · ${esc(src)}${d.allowed ? ` · allowed now: ${esc(d.allowed.join(", "))}` : ""}</div>
+ ${d.note ? `<div class="note bad">${esc(d.note)}</div>` : ""}`;
+}
+
+function layout(t) { return t ? Object.keys(t).sort().map((k) => t[k]).join("/") : "–"; }
+
+function memoryHTML(mem, win) {
+ if (!mem) return '<span class="dim">empty</span>';
+ const inc = mem.incident;
+ let html = "";
+ if (win) html += `<div class="kv"><span>skewed windows</span><b>${esc(win.skewed_windows ?? 0)} in a row</b></div>`;
+ if (inc) {
+ const b = inc.baseline;
+ html += `<div class="kv"><span>incident</span><b>#${inc.id} · ${esc(inc.phase)}</b>
+ <span>config before</span><span class="mono">${esc(inc.config_before?.mode)}, ${esc(inc.config_before?.interval_ms)}ms</span>
+ <span>baseline</span><span>${b ? `<b>${num(b.records_per_sec)} rec/s</b> · tasks/TM ${layout(b.tasks_per_tm)} · skew ${esc(b.skew)}` : '<span class="dim">not taken yet</span>'}</span>
+ ${inc.increases ? `<span>increases</span><b>${inc.increases} of 3</b>` : ""}</div>`;
+ if (inc.attempts.length) {
+ html += `<table style="margin-top:8px"><tr><th>#</th><th>action</th><th>request</th><th>took</th><th>after</th></tr>` +
+ inc.attempts.map((a, i) => `<tr><td>${i + 1}</td><td><b>${esc(a.label)}</b></td><td class="mono">${esc(a.request_id)}</td>
+ <td>${a.took_s ? a.took_s + "s" : "…"}${a.reconciled ? ' <span style="color:var(--green)">reconciled</span>' : ""}</td>
+ <td>${a.observed_after ? `${layout(a.observed_after.tasks_per_tm)} · ${num(a.observed_after.records_per_sec)} rec/s` : ""}</td></tr>`).join("") + "</table>";
+ }
+ } else {
+ html += `<div class="kv"><span>incident</span><span class="dim">none open</span></div>`;
+ }
+ if (mem.history && mem.history.length) {
+ html += `<div style="margin-top:8px"><span class="dim">closed:</span> ` + mem.history.map((h) =>
+ `<b>#${h.id} ${esc(h.outcome)}</b> <span class="dim">(${esc(h.actions.join(" → ") || "no action")}${h.baseline ? `; baseline ${num(h.baseline.records_per_sec)} rec/s` : ""})</span>`).join(" · ") + "</div>";
+ }
+ if (mem.tasks_reverted) html += '<div class="note bad">TASKS was reverted before for this job</div>';
+ if (mem.escalated) html += `<div class="note hold">escalated as <b class="mono">${esc(mem.escalated.request_id)}</b>: hands off, the agent watches but changes nothing until the job is healthy</div>`;
+ return html;
+}
+
+function timelineHTML(items) {
+ if (!items || !items.length) return '<span class="dim">waiting for the first window</span>';
+ return items.slice().reverse().map((t) => `<div class="tl"><span class="dim">${hms(t.window_end || t.at)}</span>
+ <span class="kind k-${t.kind}">${t.kind}</span>
+ <span>${esc(t.text)} ${t.kind === "observe" && t.tasks_per_tm ? `<span class="dim">· ${layout(t.tasks_per_tm)}${t.rate ? " · " + num(t.rate) + " rec/s" : ""}</span>` : ""}</span></div>`).join("");
+}
+
+function reportHTML(e) {
+ if (!e) return '<span class="dim">none</span>';
+ const lines = e.body.split("\n");
+ let html = `<div class="dim" style="margin-bottom:6px">${esc(e.title)}</div>`, table = [], list = [];
+ const inline = (s) => esc(s).replace(/`([^`]*)`/g, "<code>$1</code>");
+ const flush = () => {
+ if (list.length) { html += "<ul>" + list.map((l) => `<li>${inline(l)}</li>`).join("") + "</ul>"; list = []; }
+ if (table.length) {
+ const rows = table.filter((r) => !/^\|[-| ]+\|$/.test(r)).map((r) => r.slice(1, -1).split("|").map((c) => c.trim()));
+ html += "<table>" + rows.map((r, i) => "<tr>" + r.map((c) => i ? `<td>${inline(c)}</td>` : `<th>${inline(c)}</th>`).join("") + "</tr>").join("") + "</table>";
+ table = [];
+ }
+ };
+ for (const l of lines) {
+ if (l.startsWith("|")) { table.push(l); continue; }
+ if (l.startsWith("- ")) { list.push(l.slice(2)); continue; }
+ flush();
+ if (l.startsWith("### ")) html += `<h4>${inline(l.slice(4))}</h4>`;
+ else if (l.trim()) html += `<div>${inline(l)}</div>`;
+ }
+ flush();
+ return html;
+}
+
+function render(s) {
+ $("c-samples").textContent = num(s.counters.samples);
+ $("c-windows").textContent = num(s.counters.windows);
+ $("c-llm").textContent = num(s.counters.llm_calls);
+ $("c-effects").textContent = num(s.counters.side_effects);
+ for (const k of ["KEPT", "REVERTED", "ESCALATED"]) $("o-" + k).textContent = s.outcomes[k];
+ const a = s.agent || {};
+ $("agent").textContent = a.state + (a.restarts ? ` · restarts ${a.restarts}` : "");
+ $("agent").className = "agent-state s-" + a.state;
+
+ const h = s.hero, live = h.live || {};
+ setHTML("h-config", configChips(live.config));
+ setHTML("h-tms", live.running ? tmRows(live.tasks_per_tm, live.cpu) : `<div class="dim" style="padding:20px 0">${live.reachable ? "job not running" : "cluster restarting…"}</div>`);
+ $("h-rate").textContent = live.running ? num(h.live_rate) : "–";
+ $("h-bp").textContent = live.running ? num(live.backpressure_ms) : "–";
+ $("h-redeploys").textContent = h.redeploys;
+ setHTML("h-deploy", deployHTML(h.deployment));
+ setHTML("h-decision", decisionHTML(h.latest_decision));
+ setHTML("h-memory", memoryHTML(h.memory, h.window));
+ setHTML("h-timeline", timelineHTML(h.timeline));
+ flashOnChange("h-decision-card", h.latest_decision && h.latest_decision.at);
+ flashOnChange("h-deploy-card", h.deployment && h.deployment.request_id + h.deployment.status + (h.deployment.notes || []).length);
+
+ const f = s.fleet, fv = f.focus, w = fv.window || {};
+ const bad = f.jobs.filter((j) => j.status !== "healthy").length;
+ $("f-count").textContent = `${f.jobs.length} jobs · ${f.jobs.length - bad} healthy`;
+ setHTML("f-tiles", f.jobs.map((j) => `<div class="tile ${j.status}${j.job === f.focus_job ? " focus" : ""}">${esc(j.job.replace("fleet/", ""))}<br><span class="dim">${layout(j.tasks_per_tm)}</span></div>`).join(""));
+ const focus = f.jobs.find((j) => j.job === f.focus_job) || {};
+ setHTML("f-name", `${esc(f.focus_job)} <span class="badge sim">SIMULATED</span>`);
+ setHTML("f-config", configChips(focus.config));
+ setHTML("f-tms", w.tasks_per_tm ? tmRows(w.tasks_per_tm, w.cpu) : tmRows(focus.tasks_per_tm, null));
+ $("f-rate").textContent = num(w.records_per_sec);
+ $("f-failovers").textContent = w.recent_failovers ?? "–";
+ $("f-redeploys").textContent = fv.redeploys;
+ setHTML("f-deploy", fv.deployment ? `<div style="margin-top:10px">${deployHTML(fv.deployment)}</div>` : "");
+ setHTML("f-decision", decisionHTML(fv.latest_decision));
+ setHTML("f-memory", memoryHTML(fv.memory, fv.window));
+ setHTML("f-escalation", reportHTML(fv.escalation));
+ setHTML("f-timeline", timelineHTML(fv.timeline));
+ flashOnChange("f-decision-card", fv.latest_decision && fv.latest_decision.at);
+ flashOnChange("f-escalation-card", fv.escalation && fv.escalation.request_id);
+}
+
+async function tick() {
+ $("clock").textContent = new Date().toLocaleTimeString("en-GB");
+ try { render(await (await fetch("/api/state")).json()); } catch (e) { $("agent").textContent = "dashboard: " + e.message; }
+}
+tick();
+setInterval(tick, 1000);
+</script>
+</body>
+</html>
diff --git a/flink-stateful-ops-agent-demo/platform-sim/fleet.py b/flink-stateful-ops-agent-demo/platform-sim/fleet.py
new file mode 100644
index 0000000..4dd4f76
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/platform-sim/fleet.py
@@ -0,0 +1,142 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""SIMULATED fleet: jobs that emit the same metric samples as the real target job.
+
+All jobs are healthy except fleet/payments-sessionizer once its incident is
+triggered: it already runs with TASKS balancing, fails over twice, and comes back
+unbalanced (4/3/2). The imbalance persists whatever slot.request.max-interval is
+applied (the FLINK-38715 case), so the runbook ends in an escalation.
+"""
+import random
+import threading
+import time
+import uuid
+
+INCIDENT_JOB = "fleet/payments-sessionizer"
+HEALTHY = [
+ "ads-click-attribution", "search-query-logger", "feed-ranking-features", "jobs-recommendation",
+ "notifications-fanout", "profile-view-counter", "messaging-delivery", "fraud-signals",
+ "learning-progress", "sales-lead-scoring", "groups-activity", "events-rsvp",
+ "creator-analytics", "hashtag-trends", "connection-suggestions", "company-follow",
+ "video-watch-time", "newsletter-opens", "skills-endorsement", "premium-billing",
+]
+REDEPLOY_STEPS = [("stop-with-savepoint", 1.5), ("recreate cluster", 4.0),
+ ("submit from savepoint", 1.5), ("wait for RUNNING", 1.0)]
+# CPU of a TaskManager by its task count while the incident job is unbalanced.
+INCIDENT_CPU = {4: 0.97, 3: 0.72, 2: 0.45}
+
+
+class SimJob:
+ def __init__(self, name, rng, tasks, mode="TASKS", interval_ms=20):
+ self.name = name
+ self.rng = rng
+ self.balanced = dict(tasks)
+ self.tasks = dict(tasks)
+ self.config = {"mode": mode, "interval_ms": interval_ms}
+ self.per_task = rng.uniform(0.13, 0.2)
+ self.rate_per_task = rng.randint(800, 2500)
+ self.records_in = 0.0
+ self.job_id = uuid.uuid4().hex
+ self.failovers = 0
+ self.incident = False
+ self.down_until = 0.0
+ self.last_ts = None
+
+ def sample(self, ts):
+ s = {"job": self.name, "simulated": True, "reachable": True, "config": dict(self.config),
+ "recent_failovers": self.failovers}
+ if ts < self.down_until:
+ self.last_ts = None
+ return {**s, "running": False}
+ if self.incident:
+ cpu = {tm: round(INCIDENT_CPU.get(n, 0.6) + self.rng.uniform(-0.03, 0.02), 2)
+ for tm, n in self.tasks.items()}
+ else:
+ cpu = {tm: round(n * self.per_task + self.rng.uniform(-0.04, 0.04), 2)
+ for tm, n in self.tasks.items()}
+ rate = sum(self.tasks.values()) * self.rate_per_task * (0.62 if self.incident else 1.0)
+ if self.last_ts is not None:
+ self.records_in += rate * (ts - self.last_ts) * self.rng.uniform(0.97, 1.03)
+ self.last_ts = ts
+ return {**s, "running": True, "job_id": self.job_id, "tasks_per_tm": dict(self.tasks), "cpu": cpu,
+ "backpressure_ms": round(self.rng.uniform(580, 760) if self.incident else self.rng.uniform(0, 60)),
+ "records_in": round(self.records_in)}
+
+ def restart(self, down_s):
+ self.down_until = time.time() + down_s
+ self.job_id = uuid.uuid4().hex
+ self.records_in = 0.0
+
+
+class Fleet:
+ def __init__(self, seed=7):
+ self.lock = threading.Lock()
+ self.seed = seed
+ self.reset()
+
+ def reset(self):
+ with self.lock:
+ rng = random.Random(self.seed)
+ self.jobs = {}
+ for name in HEALTHY:
+ n = rng.choice([2, 3, 3, 4])
+ tms = rng.choice([3, 3, 4])
+ mode = rng.choice(["TASKS", "TASKS", "NONE"])
+ job = SimJob(f"fleet/{name}", rng, {f"tm-{i + 1}": n for i in range(tms)}, mode)
+ self.jobs[job.name] = job
+ self.jobs[INCIDENT_JOB] = SimJob(INCIDENT_JOB, rng, {"tm-1": 3, "tm-2": 3, "tm-3": 3}, "TASKS", 20)
+
+ def samples(self, ts):
+ with self.lock:
+ return [job.sample(ts) for job in self.jobs.values()]
+
+ def trigger_incident(self, name=INCIDENT_JOB):
+ """Two failovers, after which TASKS balancing leaves the job at 4/3/2."""
+ with self.lock:
+ job = self.jobs[name]
+ job.failovers = 2
+ job.incident = True
+ job.tasks = {"tm-1": 4, "tm-2": 3, "tm-3": 2}
+ job.restart(4)
+
+ def redeploy(self, name, mode, interval_ms, on_step):
+ """Simulated redeploy with the same steps as the real one; the imbalance survives it."""
+ with self.lock:
+ job = self.jobs[name]
+ job.restart(sum(d for _, d in REDEPLOY_STEPS) + 1)
+ for step, seconds in REDEPLOY_STEPS:
+ on_step(step)
+ time.sleep(seconds)
+ if step == "recreate cluster":
+ with self.lock:
+ job.config = {"mode": mode, "interval_ms": int(interval_ms)}
+ with self.lock:
+ if not job.incident:
+ job.tasks = dict(job.balanced)
+ return {"job_id": job.job_id, "simulated": True}
+
+ def summary(self):
+ now = time.time()
+ with self.lock:
+ out = []
+ for job in self.jobs.values():
+ counts = list(job.tasks.values())
+ status = "redeploying" if now < job.down_until else ("unbalanced" if job.incident else "healthy")
+ out.append({"job": job.name, "status": status, "config": dict(job.config),
+ "tasks_per_tm": dict(job.tasks), "task_skew": max(counts) - min(counts)})
+ return out
diff --git a/flink-stateful-ops-agent-demo/platform-sim/platform_sim.py b/flink-stateful-ops-agent-demo/platform-sim/platform_sim.py
new file mode 100644
index 0000000..d68ae3b
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/platform-sim/platform_sim.py
@@ -0,0 +1,414 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""The platform around the agent, in one process (stands in for metrics, a
+deployment API and a ticket system):
+
+- collector: every 3s, one metric sample per watched job -> Kafka `ops_metrics`
+ (the REAL target job from its REST API, plus the SIMULATED fleet)
+- deployment API: executes every redeploy request it receives, without
+ de-duplication, and logs each execution to tmp/platform/deployments.jsonl
+- escalation API: files the agent's reports to tmp/platform/escalations/
+- dashboard: http://localhost:8090, built from Kafka `ops_records` (the agent's
+ output) and the platform's own view
+"""
+import json
+import os
+import re
+import sys
+import threading
+import time
+import traceback
+import urllib.parse
+import urllib.request
+from collections import Counter, deque
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+from kafka import KafkaConsumer, KafkaProducer
+from kafka.errors import KafkaError
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parent
+sys.path.insert(0, str(ROOT / "target-cluster"))
+
+import collector # noqa: E402
+import target_ctl # noqa: E402
+from fleet import INCIDENT_JOB, Fleet # noqa: E402
+
+KAFKA = os.environ.get("KAFKA_BOOTSTRAP", "localhost:9092")
+PORT = int(os.environ.get("PLATFORM_PORT", "8090"))
+SAMPLE_INTERVAL_S = 3
+HERO = target_ctl.JOB_NAME
+DATA = ROOT / "tmp" / "platform"
+AGENT_REST = os.environ.get("AGENT_FLINK_REST", "http://localhost:8082")
+
+
+def agent_job_status():
+ """State of the latest agent job on the agent cluster, for the dashboard."""
+ try:
+ with urllib.request.urlopen(AGENT_REST + "/jobs/overview", timeout=2) as resp:
+ jobs = json.loads(resp.read())["jobs"]
+ except (OSError, ValueError, KeyError):
+ return {"state": "UNREACHABLE"}
+ if not jobs:
+ return {"state": "NO JOB"}
+ job = max(jobs, key=lambda j: j["start-time"])
+ out = {"state": job["state"], "jid": job["jid"], "name": job["name"], "start_time": job["start-time"] / 1000}
+ try:
+ url = f"{AGENT_REST}/jobs/{job['jid']}/metrics?get=numRestarts"
+ with urllib.request.urlopen(url, timeout=2) as resp:
+ values = json.loads(resp.read())
+ out["restarts"] = int(float(values[0]["value"])) if values else 0
+ except (OSError, ValueError, KeyError, IndexError):
+ pass
+ return out
+
+
+def log(msg):
+ print(time.strftime("%H:%M:%S"), msg, flush=True)
+
+
+def _safe(name):
+ return re.sub(r"[^A-Za-z0-9._-]", "_", name)
+
+
+class Platform:
+ def __init__(self):
+ self.lock = threading.RLock()
+ self.real_lock = threading.Lock()
+ self.fleet = Fleet()
+ (DATA / "escalations").mkdir(parents=True, exist_ok=True)
+ self.state_file = DATA / "state.json"
+ saved = json.loads(self.state_file.read_text()) if self.state_file.exists() else {}
+ self.ignore_before = saved.get("ignore_before", 0)
+ self._clear()
+
+ def _clear(self):
+ self.samples = 0
+ self.hero_samples = deque(maxlen=6)
+ self.deployments = {}
+ self.executions = []
+ self.escalations = {}
+ self.records = {}
+ self.order = []
+ self.duplicate_records = 0
+ self.agent = {"state": "UNKNOWN"}
+
+ def _ledger(self, event, rec):
+ with open(DATA / "deployments.jsonl", "a") as f:
+ f.write(json.dumps({"event": event, "at": time.time(), **rec}) + "\n")
+
+ # ---- collector -------------------------------------------------------
+ def collect_loop(self):
+ producer = None
+ next_t = time.time()
+ seq = 0
+ while True:
+ try:
+ if producer is None:
+ producer = KafkaProducer(
+ bootstrap_servers=KAFKA, linger_ms=20,
+ key_serializer=str.encode, value_serializer=lambda v: json.dumps(v).encode())
+ ts = time.time()
+ batch = [collector.sample(HERO, ts)] + self.fleet.samples(ts)
+ for s in batch:
+ s.update(ts=ts, seq=seq)
+ producer.send("ops_metrics", key=s["job"], value=s, timestamp_ms=int(ts * 1000))
+ producer.flush()
+ seq += 1
+ agent = agent_job_status()
+ with self.lock:
+ self.samples += len(batch)
+ self.hero_samples.append(batch[0])
+ self.agent = agent
+ except KafkaError as e:
+ log(f"collector: Kafka error ({e!r}), reconnecting")
+ producer = None
+ except Exception: # keep sampling whatever happens
+ log("collector error: " + traceback.format_exc(limit=2))
+ next_t = max(next_t + SAMPLE_INTERVAL_S, time.time() - SAMPLE_INTERVAL_S)
+ time.sleep(max(0.0, next_t - time.time()))
+
+ # ---- the agent's output ----------------------------------------------
+ def records_loop(self):
+ while True:
+ try:
+ consumer = KafkaConsumer(
+ "ops_records", bootstrap_servers=KAFKA, group_id=None, auto_offset_reset="earliest",
+ enable_auto_commit=False, value_deserializer=lambda b: json.loads(b))
+ for msg in consumer:
+ rec = msg.value
+ with self.lock:
+ if rec.get("at", 0) < self.ignore_before:
+ continue
+ if rec["id"] in self.records:
+ self.duplicate_records += 1
+ continue
+ self.records[rec["id"]] = rec
+ self.order.append(rec["id"])
+ except Exception:
+ log("records consumer error: " + traceback.format_exc(limit=2))
+ time.sleep(3)
+
+ # ---- deployment API ----------------------------------------------------
+ def create_deployment(self, body):
+ rid = body["request_id"]
+ with self.lock:
+ n = sum(1 for e in self.executions if e["request_id"] == rid) + 1
+ rec = {
+ "request_id": rid, "job": body["job"], "target": {"mode": body["mode"],
+ "interval_ms": int(body["interval_ms"])},
+ "reason": body.get("reason"), "agent_pid": body.get("agent_pid"), "status": "running",
+ "created_at": time.time(), "execution": n, "steps": [], "notes": [],
+ }
+ if n > 1:
+ rec["notes"].append(f"DUPLICATE: request {rid} executed {n} times")
+ self.deployments[rid] = rec
+ self.executions.append({"request_id": rid, "job": body["job"], "at": rec["created_at"]})
+ log(f"deploy {rid}: {rec['target']} (execution {n}, agent pid {rec['agent_pid']})")
+ self._ledger("received", {k: rec[k] for k in ("request_id", "job", "target", "agent_pid", "execution")})
+ threading.Thread(target=self._run_deployment, args=(rec,), daemon=True).start()
+ return rec
+
+ def _step(self, rec, name):
+ with self.lock:
+ if rec["steps"]:
+ rec["steps"][-1]["done"] = True
+ rec["steps"].append({"name": name, "at": time.time(), "done": False})
+
+ def _run_deployment(self, rec):
+ t0 = time.time()
+ try:
+ if rec["job"] == HERO:
+ result = self._real_redeploy(rec)
+ else:
+ result = self.fleet.redeploy(rec["job"], rec["target"]["mode"], rec["target"]["interval_ms"],
+ lambda name: self._step(rec, name))
+ with self.lock:
+ rec["steps"][-1]["done"] = True
+ rec.update(status="done", result=result, took_s=round(time.time() - t0, 1))
+ except Exception as e:
+ with self.lock:
+ rec.update(status="failed", error=f"{type(e).__name__}: {e}"[:500])
+ log(f"deploy {rec['request_id']}: {rec['status']} in {round(time.time() - t0, 1)}s")
+ self._ledger(rec["status"], {k: rec.get(k) for k in ("request_id", "job", "took_s", "result", "error")})
+
+ def _real_redeploy(self, rec):
+ """Savepoint, recreate the target cluster with the new config, restore."""
+ with self.real_lock:
+ self._step(rec, "stop-with-savepoint")
+ jid = target_ctl.running_job_id()
+ savepoint = target_ctl.stop_with_savepoint(jid) if jid else None
+ self._step(rec, "recreate cluster")
+ target_ctl.up(rec["target"]["mode"], rec["target"]["interval_ms"])
+ self._step(rec, "submit from savepoint")
+ new_jid = target_ctl.submit(savepoint)
+ self._step(rec, "wait for RUNNING")
+ target_ctl.wait_for_running(new_jid)
+ return {"savepoint": savepoint, "job_id": new_jid}
+
+ def reconciled(self, rid, body):
+ with self.lock:
+ rec = self.deployments.get(rid)
+ if rec is None:
+ return None
+ rec["reconciled"] = {"at": time.time(), "agent_pid": body.get("agent_pid")}
+ rec["notes"].append("agent recovered: its reconciler found this request and waited "
+ "for it instead of sending it again")
+ log(f"deploy {rid}: reconciled by agent pid {body.get('agent_pid')}")
+ self._ledger("reconciled", {"request_id": rid, "agent_pid": body.get("agent_pid")})
+ return rec
+
+ def inflight(self):
+ with self.lock:
+ running = [d for d in self.deployments.values() if d["status"] == "running"]
+ return running[-1] if running else None
+
+ # ---- escalation API ----------------------------------------------------
+ def create_escalation(self, body):
+ rid = body["request_id"]
+ with self.lock:
+ esc = self.escalations.get(rid) or {**body, "at": time.time(), "count": 0}
+ esc["count"] += 1
+ self.escalations[rid] = esc
+ (DATA / "escalations" / f"{_safe(rid)}.md").write_text(f"# {body['title']}\n\n{body['body']}\n")
+ log(f"escalation {rid} filed (count {esc['count']})")
+ return esc
+
+ # ---- control -------------------------------------------------------------
+ def reset(self):
+ with self.lock:
+ self._clear()
+ self.ignore_before = time.time()
+ self.state_file.write_text(json.dumps({"ignore_before": self.ignore_before}))
+ self.fleet.reset()
+ log("reset")
+
+ # ---- dashboard state -------------------------------------------------------
+ def _live_rate(self):
+ samples = [s for s in self.hero_samples if s.get("running")]
+ if not samples:
+ return None
+ same = [s for s in samples if s["job_id"] == samples[-1]["job_id"]]
+ deltas = sorted((b["records_in"] - a["records_in"]) / (b["ts"] - a["ts"])
+ for a, b in zip(same, same[1:]) if b["ts"] - a["ts"] >= 1)
+ return round(deltas[len(deltas) // 2]) if deltas else None
+
+ def _job_view(self, job, recs):
+ decisions = [r for r in recs if r["kind"] == "decision"]
+ deployments = [d for d in self.deployments.values() if d["job"] == job]
+ escalations = [e for e in self.escalations.values() if e["job"] == job]
+ return {
+ "memory": recs[-1]["memory"] if recs else None,
+ "window": recs[-1]["window"] if recs else None,
+ "timeline": [_timeline_item(r) for r in recs[-12:]],
+ "latest_decision": {k: decisions[-1].get(k) for k in
+ ("at", "decision", "source", "note", "allowed", "llm_seconds", "request_id")}
+ if decisions else None,
+ "redeploys": sum(1 for e in self.executions if e["job"] == job),
+ "deployment": deployments[-1] if deployments else None,
+ "escalation": escalations[-1] if escalations else None,
+ }
+
+ def state(self):
+ with self.lock:
+ recs = [self.records[i] for i in self.order]
+ by_job = {}
+ for r in recs:
+ by_job.setdefault(r["job"], []).append(r)
+ kinds = Counter(r["kind"] for r in recs)
+ outcomes = Counter()
+ for rs in by_job.values():
+ for h in (rs[-1].get("memory") or {}).get("history") or []:
+ outcomes[h["outcome"]] += 1
+ hero_live = self.hero_samples[-1] if self.hero_samples else None
+ return {
+ "now": time.time(),
+ "counters": {
+ "samples": self.samples,
+ "windows": kinds["observe"] + kinds["decision"],
+ "llm_calls": sum(1 for r in recs if r["kind"] == "decision" and "llm_seconds" in r),
+ "side_effects": len(self.executions) + sum(e["count"] for e in self.escalations.values()),
+ "duplicate_records": self.duplicate_records,
+ },
+ "outcomes": {k: outcomes.get(k, 0) for k in ("KEPT", "REVERTED", "ESCALATED")},
+ "agent": self.agent,
+ "hero": {"job": HERO, "live": hero_live, "live_rate": self._live_rate(),
+ **self._job_view(HERO, by_job.get(HERO, []))},
+ "fleet": {"jobs": self.fleet.summary(), "focus_job": INCIDENT_JOB,
+ "focus": self._job_view(INCIDENT_JOB, by_job.get(INCIDENT_JOB, []))},
+ }
+
+
+def _timeline_item(r):
+ w = r.get("window") or {}
+ if r["kind"] == "observe":
+ text = r.get("note") or ""
+ elif r["kind"] == "decision":
+ d = r.get("decision") or {}
+ text = f"{d.get('action')} · {d.get('runbook_step')}"
+ elif r["kind"] == "applied":
+ rc = r.get("receipt") or {}
+ text = f"redeployed {r.get('request_id')} in {rc.get('took_s')}s" + (" (reconciled)" if rc.get("reconciled") else "")
+ else:
+ text = f"escalated {r.get('request_id')}"
+ return {"at": r.get("at"), "window_end": w.get("window_end"), "kind": r["kind"], "text": text,
+ "source": r.get("source"), "skew": w.get("task_skew"), "rate": w.get("records_per_sec"),
+ "tasks_per_tm": w.get("tasks_per_tm")}
+
+
+class Handler(BaseHTTPRequestHandler):
+ platform: Platform = None
+
+ def log_message(self, *args):
+ pass
+
+ def _send(self, code, obj=None, content_type="application/json", raw=None):
+ body = raw if raw is not None else json.dumps(obj).encode()
+ self.send_response(code)
+ self.send_header("Content-Type", content_type)
+ self.send_header("Content-Length", str(len(body)))
+ self.send_header("Cache-Control", "no-store")
+ self.end_headers()
+ self.wfile.write(body)
+
+ def _parts(self):
+ path = urllib.parse.urlparse(self.path).path
+ return [urllib.parse.unquote(p) for p in path.strip("/").split("/") if p]
+
+ def _body(self):
+ n = int(self.headers.get("Content-Length") or 0)
+ return json.loads(self.rfile.read(n) or b"{}")
+
+ def do_GET(self):
+ p, plat = self._parts(), self.platform
+ if p in ([], ["index.html"]):
+ return self._send(200, content_type="text/html; charset=utf-8",
+ raw=(HERE / "dashboard.html").read_bytes())
+ if p == ["api", "state"]:
+ return self._send(200, plat.state())
+ if p == ["api", "inflight"]:
+ rec = plat.inflight()
+ return self._send(200 if rec else 404, rec or {"error": "no deployment in flight"})
+ if p == ["api", "deployments"]:
+ with plat.lock:
+ return self._send(200, list(plat.deployments.values()))
+ if len(p) == 3 and p[:2] == ["api", "deployments"]:
+ with plat.lock:
+ rec = plat.deployments.get(p[2])
+ return self._send(200 if rec else 404, rec or {"error": "not found"})
+ if len(p) == 3 and p[:2] == ["api", "escalations"]:
+ with plat.lock:
+ esc = plat.escalations.get(p[2])
+ return self._send(200 if esc else 404, esc or {"error": "not found"})
+ self._send(404, {"error": "not found"})
+
+ def do_POST(self):
+ p, plat = self._parts(), self.platform
+ try:
+ if p == ["api", "deployments"]:
+ return self._send(202, plat.create_deployment(self._body()))
+ if len(p) == 4 and p[:2] == ["api", "deployments"] and p[3] == "reconciled":
+ rec = plat.reconciled(p[2], self._body())
+ return self._send(200 if rec else 404, rec or {"error": "not found"})
+ if p == ["api", "escalations"]:
+ return self._send(201, plat.create_escalation(self._body()))
+ if p == ["api", "fleet", "incident"]:
+ plat.fleet.trigger_incident()
+ log(f"fleet incident triggered on {INCIDENT_JOB}")
+ return self._send(200, {"triggered": INCIDENT_JOB})
+ if p == ["api", "reset"]:
+ plat.reset()
+ return self._send(200, {"reset": True})
+ except (KeyError, ValueError) as e:
+ return self._send(400, {"error": str(e)})
+ self._send(404, {"error": "not found"})
+
+
+def main():
+ platform = Platform()
+ Handler.platform = platform
+ threading.Thread(target=platform.collect_loop, daemon=True, name="collector").start()
+ threading.Thread(target=platform.records_loop, daemon=True, name="records").start()
+ server = ThreadingHTTPServer(("localhost", PORT), Handler)
+ log(f"platform-sim on http://localhost:{PORT} (target {target_ctl.REST}, Kafka {KAFKA})")
+ server.serve_forever()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/flink-stateful-ops-agent-demo/skills/balanced-task-scheduling/SKILL.md b/flink-stateful-ops-agent-demo/skills/balanced-task-scheduling/SKILL.md
new file mode 100644
index 0000000..43c7c3c
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/skills/balanced-task-scheduling/SKILL.md
@@ -0,0 +1,71 @@
+---
+name: balanced-task-scheduling
+description: Runbook for a Flink job where some TaskManagers run more tasks than others and become the bottleneck (backpressure, a hot TaskManager at full CPU while others idle). Covers when to switch taskmanager.load-balance.mode to TASKS, how to verify it against a baseline, how to tune slot.request.max-interval when the balance does not hold after failovers, and when to stop and escalate.
+license: Apache-2.0
+---
+
+# Balanced Tasks Scheduling runbook (Flink 2.2+, FLIP-370)
+
+Sources:
+- https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/deployment/tasks-scheduling/balanced_tasks_scheduling/
+- https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/tasks-scheduling/balanced_tasks_scheduling/ (failover tuning note)
+
+## Background
+With the default `taskmanager.load-balance.mode: NONE`, slots are filled without
+looking at how many tasks each slot carries. When operators have different
+parallelism (for example 6 and 3), some slots hold 2 tasks and some hold 1, so one
+TaskManager can end up with 4 tasks while another has 2. If tasks are CPU-heavy,
+the TaskManager with the most tasks saturates and backpressures the whole job,
+while the others idle. Adding machines does not fix this; placement does.
+
+`TASKS` mode waits until all slot requests are known, then assigns the slots with
+the most tasks first, each to the TaskManager that currently has the fewest tasks.
+
+## Step 1: Confirm the bottleneck is real and persistent
+Only act when the imbalance persists for at least 3 consecutive observation windows:
+- task count differs between TaskManagers (skew >= 2), AND
+- the TaskManager with the most tasks is near 100% CPU / busy while another is
+ well below, AND
+- upstream tasks are backpressured.
+A single snapshot is not enough: while fewer than 3 windows show it, WAIT. The docs
+caution against enabling TASKS when you are not seeing these bottlenecks: it may
+degrade the job.
+
+## Step 2: Record a baseline, then switch to TASKS
+Before changing anything, record the baseline: tasks per TaskManager, CPU per
+TaskManager, and throughput. Then set `taskmanager.load-balance.mode: TASKS`.
+In a session cluster this is cluster configuration, so apply it by redeploying the
+cluster and restoring the job from a savepoint.
+
+## Step 3: Verify against the baseline
+After the job is RUNNING again and metrics have settled:
+- KEEP if the task skew dropped and throughput did not get worse.
+- REVERT to the previous mode if throughput is lower than the baseline.
+
+## Step 4: If the balance does not hold after failovers
+After failovers, delayed updates of the resource view can leave the placement
+less balanced than optimal. If TASKS is already on and the skew is back, increase
+`slot.request.max-interval` by 50 ms (default 20 ms, so 70, then 120, then 170)
+and re-verify after every change. A larger value also raises the risk of hitting
+`slot.request.timeout`, so make at most 3 increases.
+
+## Step 5: Stop and escalate
+If the job is still unbalanced after 3 increases, stop tuning and report it in
+FLINK-38715 (https://issues.apache.org/jira/browse/FLINK-38715). Include the
+scheduling configuration and what was observed after each attempt.
+
+## Decision table
+| Situation | Action |
+|---|---|
+| No skew or no backpressure, no open incident | NO_ACTION |
+| Skew + hot TaskManager + backpressure for fewer than 3 windows | WAIT |
+| Same for 3+ windows, mode is NONE, TASKS not tried yet | APPLY_TASKS (record baseline first) |
+| TASKS just applied, cooldown over, throughput >= baseline | KEEP |
+| TASKS just applied, cooldown over, throughput < baseline | REVERT |
+| TASKS on, skew back after failovers, fewer than 3 increases so far | INCREASE_INTERVAL by 50 ms |
+| TASKS on, still skewed after 3 increases | ESCALATE |
+
+## Guardrails
+- One change at a time; verify before the next change.
+- Never repeat an action that was already applied for the current incident.
+- Respect the cooldown after a change before judging its effect.
diff --git a/flink-stateful-ops-agent-demo/spikes/failover_trials.py b/flink-stateful-ops-agent-demo/spikes/failover_trials.py
new file mode 100644
index 0000000..c352d74
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/spikes/failover_trials.py
@@ -0,0 +1,83 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""P0 spike: does a TaskManager failover under TASKS mode leave the job unbalanced?
+
+Kills one TaskManager container per trial (rotating), restarts it after
+RESTART_DELAY_S, waits for the job to recover, and prints the resulting layout.
+
+Usage: python spikes/failover_trials.py <trials>
+"""
+import json
+import os
+import subprocess
+import sys
+import time
+
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "target-cluster"))
+import target_ctl as t # noqa: E402
+
+RESTART_DELAY_S = float(os.environ.get("RESTART_DELAY_S", "5"))
+SETTLE_S = float(os.environ.get("SETTLE_S", "20"))
+
+
+def job_state(jid):
+ return t._get(f"/jobs/{jid}")["state"]
+
+
+def num_restarts(jid):
+ return int(t._metric(f"/jobs/{jid}/metrics", "numRestarts") or 0)
+
+
+def trial(i):
+ jid = t.running_job_id()
+ victim = f"target-tm-{i % t.NUM_TASKMANAGERS + 1}"
+ restarts_before = num_restarts(jid)
+ killed_at = time.time()
+ subprocess.run(["docker", "kill", victim], check=True, stdout=subprocess.DEVNULL)
+ time.sleep(RESTART_DELAY_S)
+ subprocess.run(["docker", "start", victim], check=True, stdout=subprocess.DEVNULL)
+ deadline = time.time() + 300
+ while time.time() < deadline:
+ if num_restarts(jid) > restarts_before:
+ try:
+ t.wait_for_running(jid, timeout_s=5)
+ break
+ except TimeoutError:
+ pass
+ time.sleep(1)
+ recovered_s = round(time.time() - killed_at, 1)
+ time.sleep(SETTLE_S)
+ t.refresh_metrics()
+ tasks = t.layout(jid)
+ counts = {tm: len(ts) for tm, ts in tasks.items()}
+ return {
+ "trial": i,
+ "victim": victim,
+ "config": t.cluster_config(),
+ "state": job_state(jid),
+ "restarts": num_restarts(jid) - restarts_before,
+ "recovered_s": recovered_s,
+ "tasks_per_tm": counts,
+ "skew": max(counts.values()) - min(counts.values()),
+ "tasks": {tm: [x["task"] for x in ts] for tm, ts in tasks.items()},
+ }
+
+
+if __name__ == "__main__":
+ for i in range(int(sys.argv[1]) if len(sys.argv) > 1 else 3):
+ print(json.dumps(trial(i)), flush=True)
diff --git a/flink-stateful-ops-agent-demo/spikes/llm_spike.py b/flink-stateful-ops-agent-demo/spikes/llm_spike.py
new file mode 100644
index 0000000..a790141
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/spikes/llm_spike.py
@@ -0,0 +1,253 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""P0 spike: can a small local model follow the runbook and return a valid Decision?
+
+Runs 7 runbook scenarios through a flink-agents agent on a PyFlink mini-cluster,
+in two variants:
+ skill the runbook is a Skill the model must load with the load_skill tool
+ inline the runbook text is in the system prompt
+Each result (action, expected action, latency, retries) is appended to OUT.
+
+Usage: cd spikes && python -c 'import llm_spike; llm_spike.main()' [skill|inline|both] [repetitions]
+"""
+import json
+import os
+import sys
+import sysconfig
+import time
+from pathlib import Path
+from typing import Literal
+
+from pydantic import BaseModel
+from pyflink.common import Configuration
+from pyflink.datastream import StreamExecutionEnvironment
+
+from flink_agents.api.agents.agent import STRUCTURED_OUTPUT, Agent
+from flink_agents.api.agents.types import OutputSchema
+from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.core_options import AgentConfigOptions, AgentExecutionOptions
+from flink_agents.api.decorators import action, chat_model_connection, chat_model_setup, skills
+from flink_agents.api.events.chat_event import ChatRequestEvent, ChatResponseEvent
+from flink_agents.api.events.event import Event, InputEvent
+from flink_agents.api.events.event_type import EventType
+from flink_agents.api.execution_environment import AgentsExecutionEnvironment
+from flink_agents.api.resource import ResourceDescriptor, ResourceName
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.skills import Skills
+
+MODULE_DIR = Path(__file__).resolve().parent.parent
+SKILLS_DIR = MODULE_DIR / "skills"
+RUNBOOK = (SKILLS_DIR / "balanced-task-scheduling" / "SKILL.md").read_text()
+OUT = os.environ.get("LLM_SPIKE_OUT", "/tmp/stateful_ops_spike/llm_spike.jsonl")
+EVENT_LOG_DIR = "/tmp/stateful_ops_spike/eventlog"
+MODEL = os.environ.get("OLLAMA_MODEL", "qwen3:8b")
+THINK = os.environ.get("OLLAMA_THINK", "false") == "true"
+# native: output_schema (Ollama's JSON format; blocks tool calls). prompt: parse the reply ourselves.
+SCHEMA = os.environ.get("SCHEMA", "native")
+
+
+def parse_decision(response) -> "Decision":
+ decision = response.extra_args.get(STRUCTURED_OUTPUT)
+ if decision is None:
+ text = response.text.strip()
+ decision = text[text.find("{"): text.rfind("}") + 1]
+ return Decision.model_validate_json(decision)
+ return decision if isinstance(decision, Decision) else Decision.model_validate(decision)
+
+Action = Literal["WAIT", "APPLY_TASKS", "KEEP", "REVERT", "INCREASE_INTERVAL", "ESCALATE", "NO_ACTION"]
+
+
+class Decision(BaseModel):
+ runbook_step: str
+ rationale: str
+ action: Action
+ interval_ms: int | None = None
+
+
+OUTPUT_FORMAT = (
+ "Decide the single next step for this job. Reply with ONLY a JSON object, no prose, "
+ "with the fields in this order:\n"
+ '{"runbook_step": "<the matching row of the runbook decision table>", '
+ '"rationale": "<one sentence: which facts from memory and the observation match that row>", '
+ '"action": "WAIT" | "APPLY_TASKS" | "KEEP" | "REVERT" | "INCREASE_INTERVAL" | '
+ '"ESCALATE" | "NO_ACTION", "interval_ms": <new slot.request.max-interval in ms, '
+ "only for INCREASE_INTERVAL, else null>}"
+)
+SYSTEM_SKILL = (
+ "You are an on-call operations agent for Apache Flink jobs. The memory section is "
+ "what you observed and did earlier for this job. Load the balanced-task-scheduling "
+ "skill first and follow it strictly.\n" + OUTPUT_FORMAT
+)
+SYSTEM_INLINE = (
+ "You are an on-call operations agent for Apache Flink jobs. The memory section is "
+ "what you observed and did earlier for this job. Follow this runbook strictly:\n\n"
+ + RUNBOOK + "\n\n" + OUTPUT_FORMAT
+)
+
+# (name, expected action, memory, latest observation)
+SCENARIOS = [
+ ("first-window", "WAIT",
+ {"incident": None, "windows_with_skew": 1, "config": {"mode": "NONE", "interval_ms": 20}},
+ {"tasks_per_tm": {"tm-1": 4, "tm-2": 3, "tm-3": 2}, "cpu": {"tm-1": 1.0, "tm-2": 0.68, "tm-3": 0.27},
+ "backpressured": True, "records_per_sec": 17000}),
+ ("persistent-skew", "APPLY_TASKS",
+ {"incident": "open", "windows_with_skew": 3, "config": {"mode": "NONE", "interval_ms": 20},
+ "baseline": None, "actions": []},
+ {"tasks_per_tm": {"tm-1": 4, "tm-2": 3, "tm-3": 2}, "cpu": {"tm-1": 1.0, "tm-2": 0.68, "tm-3": 0.27},
+ "backpressured": True, "records_per_sec": 17000}),
+ ("verify-improved", "KEEP",
+ {"incident": "open", "config": {"mode": "TASKS", "interval_ms": 20},
+ "baseline": {"skew": 2, "records_per_sec": 17000}, "actions": ["APPLY_TASKS"],
+ "cooldown_over": True},
+ {"tasks_per_tm": {"tm-1": 3, "tm-2": 3, "tm-3": 3}, "cpu": {"tm-1": 1.0, "tm-2": 1.0, "tm-3": 1.0},
+ "backpressured": False, "records_per_sec": 28000}),
+ ("verify-worse", "REVERT",
+ {"incident": "open", "config": {"mode": "TASKS", "interval_ms": 20},
+ "baseline": {"skew": 2, "records_per_sec": 17000}, "actions": ["APPLY_TASKS"],
+ "cooldown_over": True},
+ {"tasks_per_tm": {"tm-1": 3, "tm-2": 3, "tm-3": 3}, "cpu": {"tm-1": 0.6, "tm-2": 0.6, "tm-3": 0.6},
+ "backpressured": False, "records_per_sec": 12000}),
+ ("unbalanced-after-failover", "INCREASE_INTERVAL",
+ {"incident": "open", "config": {"mode": "TASKS", "interval_ms": 20},
+ "baseline": {"skew": 2, "records_per_sec": 17000}, "actions": ["APPLY_TASKS", "KEEP"],
+ "since_last_action": "2 failovers; skew back for 3 windows", "cooldown_over": True},
+ {"tasks_per_tm": {"tm-1": 4, "tm-2": 3, "tm-3": 2}, "cpu": {"tm-1": 1.0, "tm-2": 0.7, "tm-3": 0.3},
+ "backpressured": True, "records_per_sec": 18000}),
+ ("increases-exhausted", "ESCALATE",
+ {"incident": "open", "config": {"mode": "TASKS", "interval_ms": 170},
+ "baseline": {"skew": 2, "records_per_sec": 17000},
+ "actions": ["APPLY_TASKS", "KEEP", "INCREASE_INTERVAL 70", "INCREASE_INTERVAL 120",
+ "INCREASE_INTERVAL 170"],
+ "after_each_increase": "still skew 2 after failover", "cooldown_over": True},
+ {"tasks_per_tm": {"tm-1": 4, "tm-2": 3, "tm-3": 2}, "cpu": {"tm-1": 1.0, "tm-2": 0.7, "tm-3": 0.3},
+ "backpressured": True, "records_per_sec": 18000}),
+ ("healthy", "NO_ACTION",
+ {"incident": None, "windows_with_skew": 0, "config": {"mode": "NONE", "interval_ms": 20}},
+ {"tasks_per_tm": {"tm-1": 3, "tm-2": 3, "tm-3": 3}, "cpu": {"tm-1": 0.4, "tm-2": 0.4, "tm-3": 0.4},
+ "backpressured": False, "records_per_sec": 9000}),
+]
+
+
+class SpikeLlmAgent(Agent):
+ @chat_model_connection
+ @staticmethod
+ def ollama() -> ResourceDescriptor:
+ return ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_CONNECTION,
+ base_url="http://localhost:11434",
+ request_timeout=180.0,
+ )
+
+ @chat_model_setup
+ @staticmethod
+ def with_skill() -> ResourceDescriptor:
+ return ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_SETUP, connection="ollama", model=MODEL,
+ temperature=0.0, num_ctx=8192, think=THINK, keep_alive="30m",
+ skills=["balanced-task-scheduling"],
+ )
+
+ @chat_model_setup
+ @staticmethod
+ def inline() -> ResourceDescriptor:
+ return ResourceDescriptor(
+ clazz=ResourceName.ChatModel.OLLAMA_SETUP, connection="ollama", model=MODEL,
+ temperature=0.0, num_ctx=8192, think=THINK, keep_alive="30m",
+ )
+
+ @skills
+ @staticmethod
+ def runbooks() -> Skills:
+ return Skills.from_local_dir(str(SKILLS_DIR))
+
+ @action(EventType.InputEvent)
+ @staticmethod
+ def decide(event: Event, ctx: RunnerContext) -> None:
+ case = InputEvent.from_event(event).input
+ ctx.sensory_memory.set("case", case)
+ ctx.sensory_memory.set("t0", time.time())
+ system = SYSTEM_SKILL if case["variant"] == "skill" else SYSTEM_INLINE
+ user = (
+ f"Job: ClickstreamEnrichment\nMemory: {json.dumps(case['memory'])}\n"
+ f"Latest observation: {json.dumps(case['observation'])}"
+ )
+ ctx.send_event(
+ ChatRequestEvent(
+ model="with_skill" if case["variant"] == "skill" else "inline",
+ messages=[ChatMessage.of(MessageRole.SYSTEM, system), ChatMessage.of(MessageRole.USER, user)],
+ output_schema=OutputSchema(output_schema=Decision) if SCHEMA == "native" else None,
+ )
+ )
+
+ @action(EventType.ChatResponseEvent)
+ @staticmethod
+ def record(event: Event, ctx: RunnerContext) -> None:
+ resp = ChatResponseEvent.from_event(event)
+ case = ctx.sensory_memory.get("case")
+ row = {
+ "variant": case["variant"] + ("+think" if THINK else ""), "rep": case["rep"], "scenario": case["name"],
+ "expected": case["expected"], "latency_s": round(time.time() - ctx.sensory_memory.get("t0"), 1),
+ "retries": resp.retry_count, "ok": resp.is_success, "schema": SCHEMA,
+ }
+ decision = None
+ if resp.is_success:
+ try:
+ decision = parse_decision(resp.response)
+ except ValueError as e:
+ row.update(error=f"parse: {e}"[:300], correct=False, raw=resp.response.text[:300])
+ else:
+ row.update(error=resp.error[:300], correct=False)
+ if decision is not None:
+ row.update(action=decision.action, interval_ms=decision.interval_ms,
+ step=decision.runbook_step, rationale=decision.rationale)
+ row["correct"] = decision.action == case["expected"]
+ with open(OUT, "a") as f:
+ f.write(json.dumps(row) + "\n")
+
+
+def main():
+ variants = ["skill", "inline"] if len(sys.argv) < 2 or sys.argv[1] == "both" else [sys.argv[1]]
+ reps = int(sys.argv[2]) if len(sys.argv) > 2 else 1
+ cases = [
+ {"variant": v, "rep": r, "name": n, "expected": e, "memory": m, "observation": o}
+ for v in variants for r in range(reps) for (n, e, m, o) in SCENARIOS
+ ]
+ config = Configuration()
+ config.set_string("python.pythonpath", f"{sysconfig.get_paths()['purelib']}:{Path(__file__).resolve().parent}")
+ config.set_string("python.executable", sys.executable)
+ env = StreamExecutionEnvironment.get_execution_environment(config)
+ env.set_parallelism(1)
+ agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env)
+ agents_env.get_config().set(AgentExecutionOptions.MAX_RETRIES, 2)
+ # Serialize LLM calls so latency_s is per call, not time queued in Ollama.
+ agents_env.get_config().set(AgentExecutionOptions.CHAT_ASYNC, False)
+ agents_env.get_config().set(AgentConfigOptions.BASE_LOG_DIR, EVENT_LOG_DIR)
+ (
+ agents_env.from_datastream(
+ input=env.from_collection(cases), key_selector=lambda c: f"{c['variant']}-{c['rep']}-{c['name']}"
+ )
+ .apply(SpikeLlmAgent())
+ .to_datastream()
+ .print()
+ )
+ agents_env.execute()
+
+
+if __name__ == "__main__":
+ sys.exit("Run via: cd spikes && python -c 'import llm_spike; llm_spike.main()' [args] "
+ "(the Python workers import the agent class by module name, not from __main__)")
diff --git a/flink-stateful-ops-agent-demo/spikes/skew_trials.py b/flink-stateful-ops-agent-demo/spikes/skew_trials.py
new file mode 100644
index 0000000..e01e548
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/spikes/skew_trials.py
@@ -0,0 +1,51 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""P0 skew spike: alternate redeploys between scheduling modes and record layout + load."""
+import json
+import os
+import sys
+import time
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "target-cluster"))
+import target_ctl as t # noqa: E402
+
+WARMUP_S = int(os.environ.get("WARMUP_S", "60"))
+
+
+def summary(label, took):
+ s = t.status()
+ tms = s["taskmanagers"]
+ row = {
+ "label": label,
+ "redeploy_s": round(took, 1),
+ "config": s["config"],
+ "tasks_per_tm": {k: v["num_tasks"] for k, v in tms.items()},
+ "cpu_per_tm": {k: round(v["cpu_load"] or 0, 2) for k, v in tms.items()},
+ "skew": s["task_skew"],
+ "rps": round(s["records_per_sec"] or 0),
+ }
+ print(json.dumps(row), flush=True)
+ return row
+
+
+for mode in sys.argv[1:]:
+ start = time.time()
+ t.redeploy(mode)
+ took = time.time() - start
+ time.sleep(WARMUP_S)
+ summary(mode, took)
diff --git a/flink-stateful-ops-agent-demo/spikes/spike_agent.py b/flink-stateful-ops-agent-demo/spikes/spike_agent.py
new file mode 100644
index 0000000..39b582b
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/spikes/spike_agent.py
@@ -0,0 +1,120 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""P0 spike: short-term memory across runs + durable side effects across a crash.
+
+Each input {"job": "...", "seq": N, "reconcile": true|false,
+"apply_secs": A, "observe_secs": O} for one job (durations default to 20s):
+ 1. reads a per-job run counter from short-term memory (keyed state),
+ 2. APPLY: durable side effect (writes a ledger line, then takes APPLY_SECONDS,
+ like a redeploy that is in flight); optionally with a reconciler,
+ 3. OBSERVE: durable slow step (OBSERVE_SECONDS, like waiting for metrics),
+ 4. updates memory and emits the result.
+Both slow steps use durable_execute_async, so other jobs (keys) can proceed
+on the same subtask meanwhile.
+
+Chaos checks (kill -9 the TaskManager whose pid is in the ledger line):
+ - kill during APPLY with reconcile=true -> reconciler finds the ledger line,
+ no second ledger line for (job, seq)
+ - kill during APPLY with reconcile=false -> the side effect runs again
+ (baseline: what a naive retry does)
+ - kill during OBSERVE -> APPLY is replayed from the action store, not re-run
+ - in all cases the run counter continues -> memory survived
+"""
+import json
+import os
+import time
+
+from flink_agents.api.agents.agent import Agent
+from flink_agents.api.decorators import action
+from flink_agents.api.events.event import Event, InputEvent, OutputEvent
+from flink_agents.api.runner_context import RunnerContext
+
+OUT_DIR = os.environ.get("SPIKE_DIR", "/tmp/stateful_ops_spike")
+LEDGER = os.path.join(OUT_DIR, "side_effect_ledger.log")
+RECONCILE_LOG = os.path.join(OUT_DIR, "reconcile.log")
+APPLY_SECONDS = 20
+OBSERVE_SECONDS = 20
+
+
+def _log(path: str, line: str) -> None:
+ os.makedirs(OUT_DIR, exist_ok=True)
+ with open(path, "a") as f:
+ f.write(f"{time.strftime('%H:%M:%S')} {line}\n")
+
+
+def _apply(job: str, seq: int, run: int, secs: int = APPLY_SECONDS) -> str:
+ receipt = f"job={job} seq={seq} run={run} pid={os.getpid()}"
+ _log(LEDGER, receipt)
+ time.sleep(secs)
+ return receipt
+
+
+def _reconcile(job: str, seq: int, run: int, secs: int = APPLY_SECONDS) -> str:
+ """Recovery-only: did the side effect already happen before the crash?"""
+ tag = f"job={job} seq={seq} "
+ if os.path.exists(LEDGER):
+ with open(LEDGER) as f:
+ for line in f:
+ if tag in line:
+ _log(RECONCILE_LOG, f"{tag}found existing -> skip re-apply")
+ return line.strip().split(" ", 1)[1]
+ _log(RECONCILE_LOG, f"{tag}not found -> apply now")
+ return _apply(job, seq, run, secs)
+
+
+def _observe(job: str, seq: int, secs: int = OBSERVE_SECONDS) -> str:
+ time.sleep(secs)
+ return "healthy"
+
+
+class SpikeAgent(Agent):
+ @action(InputEvent.EVENT_TYPE)
+ @staticmethod
+ async def handle(event: Event, ctx: RunnerContext) -> None:
+ raw = InputEvent.from_event(event).input
+ data = raw if isinstance(raw, dict) else json.loads(raw)
+ job, seq = data["job"], int(data["seq"])
+ use_reconciler = bool(data.get("reconcile", True))
+ apply_secs = int(data.get("apply_secs", APPLY_SECONDS))
+ observe_secs = int(data.get("observe_secs", OBSERVE_SECONDS))
+ stm = ctx.short_term_memory
+ runs = (stm.get("runs") or 0) + 1
+ history = stm.get("history") or []
+ started = time.strftime("%H:%M:%S")
+
+ if use_reconciler:
+ receipt = await ctx.durable_execute_async(
+ _apply, job, seq, runs, apply_secs,
+ reconciler=lambda: _reconcile(job, seq, runs, apply_secs),
+ )
+ else:
+ receipt = await ctx.durable_execute_async(_apply, job, seq, runs, apply_secs)
+ verdict = await ctx.durable_execute_async(_observe, job, seq, observe_secs)
+
+ history.append(seq)
+ stm.set("runs", runs)
+ stm.set("history", history)
+ ctx.send_event(
+ OutputEvent(
+ output=json.dumps(
+ {"job": job, "seq": seq, "runs": runs, "history": history,
+ "receipt": receipt, "verdict": verdict, "started": started,
+ "finished": time.strftime("%H:%M:%S")}
+ )
+ )
+ )
diff --git a/flink-stateful-ops-agent-demo/spikes/spike_main.py b/flink-stateful-ops-agent-demo/spikes/spike_main.py
new file mode 100644
index 0000000..40e718a
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/spikes/spike_main.py
@@ -0,0 +1,70 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""P0 spike job entry point (see spike_agent.py)."""
+import json
+
+from flink_agents.api.execution_environment import AgentsExecutionEnvironment
+from pyflink.common import SimpleStringSchema, WatermarkStrategy
+from pyflink.common.typeinfo import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.datastream.connectors import DeliveryGuarantee
+from pyflink.datastream.connectors.kafka import (
+ KafkaOffsetsInitializer,
+ KafkaRecordSerializationSchema,
+ KafkaSink,
+ KafkaSource,
+)
+
+from spike_agent import SpikeAgent
+
+if __name__ == "__main__":
+ env = StreamExecutionEnvironment.get_execution_environment()
+ env.set_parallelism(2)
+ source = (
+ KafkaSource.builder()
+ .set_bootstrap_servers("localhost:9092")
+ .set_topics("spike_in7")
+ .set_group_id("spike")
+ .set_starting_offsets(KafkaOffsetsInitializer.earliest())
+ .set_value_only_deserializer(SimpleStringSchema())
+ .build()
+ )
+ # Python map -> elements become pickled Python objects, as the agent expects.
+ src = env.from_source(
+ source, WatermarkStrategy.no_watermarks(), "spike_in7"
+ ).map(lambda s: s)
+ agent_env = AgentsExecutionEnvironment.get_execution_environment(env)
+ out = (
+ agent_env.from_datastream(src, key_selector=lambda s: json.loads(s)["job"])
+ .apply(SpikeAgent())
+ .to_datastream()
+ .map(lambda o: str(o), output_type=Types.STRING())
+ )
+ out.sink_to(
+ KafkaSink.builder()
+ .set_bootstrap_servers("localhost:9092")
+ .set_record_serializer(
+ KafkaRecordSerializationSchema.builder()
+ .set_topic("spike_out7")
+ .set_value_serialization_schema(SimpleStringSchema())
+ .build()
+ )
+ .set_delivery_guarantee(DeliveryGuarantee.AT_LEAST_ONCE)
+ .build()
+ )
+ agent_env.execute()
diff --git a/flink-stateful-ops-agent-demo/target-cluster/docker-compose.yml b/flink-stateful-ops-agent-demo/target-cluster/docker-compose.yml
new file mode 100644
index 0000000..8f51afa
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/target-cluster/docker-compose.yml
@@ -0,0 +1,80 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# The TARGET Flink cluster the agent operates on: 1 JobManager + 3 TaskManagers,
+# 2 slots and 1 CPU each, so the TaskManager hosting the most tasks is the
+# bottleneck. In a Flink 2.2 session cluster the scheduling options are read
+# from the cluster config, so changing them means redeploying the cluster
+# (in production: a FlinkDeployment spec change).
+# LOAD_BALANCE_MODE taskmanager.load-balance.mode (NONE | SLOTS | TASKS)
+# SLOT_REQUEST_MAX_INTERVAL slot.request.max-interval in ms (Flink default 20)
+# TM_CPUS CPU cap per TaskManager (default 1)
+name: stateful-ops-target
+
+x-flink-config: &flink-config
+ FLINK_PROPERTIES: |
+ jobmanager.rpc.address: jobmanager
+ jobmanager.memory.process.size: 1024m
+ taskmanager.memory.process.size: 1280m
+ taskmanager.numberOfTaskSlots: 2
+ taskmanager.load-balance.mode: ${LOAD_BALANCE_MODE:-NONE}
+ slot.request.max-interval: ${SLOT_REQUEST_MAX_INTERVAL:-20}ms
+ execution.checkpointing.interval: 10s
+ state.checkpoints.dir: file:///flink-data/checkpoints
+ state.savepoints.dir: file:///flink-data/savepoints
+ metrics.latency.interval: 0
+ metrics.fetcher.update-interval: 2000
+
+x-taskmanager: &taskmanager
+ image: flink:2.2.1-scala_2.12-java17
+ depends_on: [jobmanager]
+ # Hard CPU cap per TaskManager. Use TM_CPUS=0.5 while developing to keep the
+ # laptop cool; the skew shows at any cap because each TaskManager is CPU-bound.
+ cpus: ${TM_CPUS:-1}
+ environment: *flink-config
+ volumes:
+ - ../tmp/target-data:/flink-data
+
+services:
+ jobmanager:
+ image: flink:2.2.1-scala_2.12-java17
+ container_name: target-jobmanager
+ hostname: jobmanager
+ command: jobmanager
+ ports:
+ - "8081:8081"
+ environment: *flink-config
+ volumes:
+ - ../tmp/target-data:/flink-data
+
+ taskmanager-1:
+ <<: *taskmanager
+ container_name: target-tm-1
+ hostname: tm-1
+ command: ["taskmanager", "-Dtaskmanager.resource-id=tm-1"]
+
+ taskmanager-2:
+ <<: *taskmanager
+ container_name: target-tm-2
+ hostname: tm-2
+ command: ["taskmanager", "-Dtaskmanager.resource-id=tm-2"]
+
+ taskmanager-3:
+ <<: *taskmanager
+ container_name: target-tm-3
+ hostname: tm-3
+ command: ["taskmanager", "-Dtaskmanager.resource-id=tm-3"]
diff --git a/flink-stateful-ops-agent-demo/target-cluster/target_ctl.py b/flink-stateful-ops-agent-demo/target-cluster/target_ctl.py
new file mode 100644
index 0000000..3eee96b
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/target-cluster/target_ctl.py
@@ -0,0 +1,309 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+"""Observe and reconfigure the TARGET Flink cluster (stdlib only).
+
+Used by the demo scripts and, later, by the agent's tools:
+ status cluster config + per-TaskManager task layout and load
+ submit [--savepoint PATH] upload and run the ClickstreamEnrichment job
+ redeploy --mode M [--interval MS]
+ stop-with-savepoint, recreate the cluster with the
+ new scheduling config, resubmit from the savepoint
+ down stop the cluster
+"""
+import argparse
+import json
+import os
+import subprocess
+import sys
+import time
+import urllib.request
+import uuid
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+COMPOSE_FILE = os.path.join(HERE, "docker-compose.yml")
+JOB_JAR = os.path.join(
+ HERE, "..", "target-jobs", "clickstream-enrichment", "target", "clickstream-enrichment.jar"
+)
+REST = os.environ.get("TARGET_FLINK_REST", "http://localhost:8081")
+JOB_NAME = "ClickstreamEnrichment"
+NUM_TASKMANAGERS = 3
+# Matches metrics.fetcher.update-interval in docker-compose.yml.
+METRICS_FETCH_INTERVAL_S = 2.0
+# Input rate for the whole job, records/s (0 = unbounded). It sits between the job's capacity
+# with skewed placement (~7.5k/s at TM_CPUS=0.5) and with balanced placement (~12k/s), so the job
+# falls behind and backpressures only while placement is skewed; once balanced it has headroom and
+# the Flink UI stops showing it busy. Capacity scales with TM_CPUS.
+JOB_RATE = int(os.environ.get("TARGET_RATE") or 19_000 * float(os.environ.get("TM_CPUS") or 1))
+
+
+def _request(method, path, body=None, headers=None, timeout=30):
+ data = json.dumps(body).encode() if isinstance(body, (dict, list)) else body
+ hdrs = {"Content-Type": "application/json"} if isinstance(body, (dict, list)) else {}
+ hdrs.update(headers or {})
+ req = urllib.request.Request(REST + path, data=data, method=method, headers=hdrs)
+ with urllib.request.urlopen(req, timeout=timeout) as resp:
+ raw = resp.read()
+ return json.loads(raw) if raw else {}
+
+
+def _get(path):
+ return _request("GET", path)
+
+
+def cluster_config():
+ keys = ("taskmanager.load-balance.mode", "slot.request.max-interval")
+ entries = {e["key"]: e["value"] for e in _get("/jobmanager/config")}
+ return {k: entries.get(k) for k in keys}
+
+
+def running_job_id(name=JOB_NAME):
+ for job in _get("/jobs/overview")["jobs"]:
+ if job["name"] == name and job["state"] == "RUNNING":
+ return job["jid"]
+ return None
+
+
+def _metric(path, name):
+ values = _get(f"{path}?get={name}")
+ return float(values[0]["value"]) if values else None
+
+
+def layout(jid):
+ """Per TaskManager: hosted subtasks with their busy/backpressured ms per second."""
+ per_tm = {}
+ for vertex in _get(f"/jobs/{jid}")["vertices"]:
+ vid = vertex["id"]
+ parts = vertex["name"].split(" -> ")
+ vname = next((p for p in parts if not p.startswith("Source:")), parts[0])
+ for st in _get(f"/jobs/{jid}/vertices/{vid}")["subtasks"]:
+ base = f"/jobs/{jid}/vertices/{vid}/subtasks/{st['subtask']}/metrics"
+ per_tm.setdefault(st["taskmanager-id"], []).append(
+ {
+ "task": f"{vname}#{st['subtask']}",
+ "status": st["status"],
+ "busy_ms": _metric(base, "busyTimeMsPerSecond"),
+ "backpressured_ms": _metric(base, "backPressuredTimeMsPerSecond"),
+ }
+ )
+ return dict(sorted(per_tm.items()))
+
+
+def throughput(jid):
+ """Records/s entering the last vertex (SessionScore), summed over subtasks.
+
+ Flink's per-second meter averages over 60s, so it lags a redeploy by about a minute;
+ use measure_throughput() for a fresh reading.
+ """
+ vertices = _get(f"/jobs/{jid}")["vertices"]
+ last = vertices[-1]["id"]
+ values = _get(
+ f"/jobs/{jid}/vertices/{last}/subtasks/metrics?get=numRecordsInPerSecond&agg=sum"
+ )
+ return float(values[0]["sum"]) if values else None
+
+
+def _records_in(jid, vid):
+ values = _get(f"/jobs/{jid}/vertices/{vid}/subtasks/metrics?get=numRecordsIn&agg=sum")
+ return float(values[0]["sum"]) if values else 0.0
+
+
+def refresh_metrics():
+ """Make the next metric reads current; returns when the fetched snapshot was taken.
+
+ The JobManager's REST metric store refreshes asynchronously: a request triggers a
+ fetch (at most every metrics.fetcher.update-interval) but is answered from the
+ previous snapshot. A cold JobManager therefore answers its first query with nothing.
+ """
+ triggered = time.time()
+ _get("/jobmanager/metrics?get=Status.JVM.CPU.Load")
+ time.sleep(METRICS_FETCH_INTERVAL_S + 0.5)
+ return triggered
+
+
+def measure_throughput(jid, seconds=20):
+ """Records/s into the last vertex over the next `seconds`, from counter deltas."""
+ last = _get(f"/jobs/{jid}")["vertices"][-1]["id"]
+ start_time = refresh_metrics()
+ start_count = _records_in(jid, last)
+ time.sleep(seconds)
+ end_time = refresh_metrics()
+ return (_records_in(jid, last) - start_count) / (end_time - start_time)
+
+
+def tm_cpu():
+ """JVM CPU load per TaskManager (container-aware: 1.0 = its one CPU is saturated)."""
+ out = {}
+ for tm in _get("/taskmanagers")["taskmanagers"]:
+ out[tm["id"]] = _metric(f"/taskmanagers/{tm['id']}/metrics", "Status.JVM.CPU.Load")
+ return dict(sorted(out.items()))
+
+
+def status():
+ jid = running_job_id()
+ result = {"config": cluster_config(), "job_id": jid}
+ if jid:
+ refresh_metrics()
+ tasks = layout(jid)
+ cpu = tm_cpu()
+ result["taskmanagers"] = {
+ tm: {
+ "num_tasks": len(ts),
+ "cpu_load": cpu.get(tm),
+ "tasks": [t["task"] for t in ts],
+ "max_busy_ms": max((t["busy_ms"] or 0) for t in ts),
+ }
+ for tm, ts in tasks.items()
+ }
+ counts = [v["num_tasks"] for v in result["taskmanagers"].values()]
+ result["task_skew"] = max(counts) - min(counts)
+ result["records_per_sec"] = throughput(jid)
+ return result
+
+
+def _upload_jar(path):
+ boundary = uuid.uuid4().hex
+ with open(path, "rb") as f:
+ content = f.read()
+ body = (
+ f"--{boundary}\r\nContent-Disposition: form-data; name=\"jarfile\"; "
+ f"filename=\"{os.path.basename(path)}\"\r\nContent-Type: application/x-java-archive\r\n\r\n"
+ ).encode() + content + f"\r\n--{boundary}--\r\n".encode()
+ resp = _request(
+ "POST",
+ "/jars/upload",
+ body,
+ {"Content-Type": f"multipart/form-data; boundary={boundary}"},
+ timeout=120,
+ )
+ return resp["filename"].split("/")[-1]
+
+
+def submit(savepoint=None, program_args=None):
+ jar_id = _upload_jar(JOB_JAR)
+ args = list(program_args or [])
+ if "--rate" not in args and JOB_RATE > 0:
+ args += ["--rate", str(JOB_RATE)]
+ body = {"programArgsList": args}
+ if savepoint:
+ body["savepointPath"] = savepoint
+ return _request("POST", f"/jars/{jar_id}/run", body, timeout=120)["jobid"]
+
+
+def stop_with_savepoint(jid, timeout_s=120):
+ trigger = _request("POST", f"/jobs/{jid}/stop", {"drain": False})["request-id"]
+ deadline = time.time() + timeout_s
+ while time.time() < deadline:
+ resp = _get(f"/jobs/{jid}/savepoints/{trigger}")
+ if resp["status"]["id"] == "COMPLETED":
+ op = resp["operation"]
+ if "failure-cause" in op:
+ raise RuntimeError(op["failure-cause"].get("stack-trace", "savepoint failed")[:2000])
+ return op["location"]
+ time.sleep(1)
+ raise TimeoutError(f"stop-with-savepoint of {jid} did not finish in {timeout_s}s")
+
+
+def wait_for_cluster(timeout_s=120):
+ deadline = time.time() + timeout_s
+ while time.time() < deadline:
+ try:
+ if len(_get("/taskmanagers")["taskmanagers"]) >= NUM_TASKMANAGERS:
+ return
+ except OSError:
+ pass
+ time.sleep(1)
+ raise TimeoutError("target cluster did not come up")
+
+
+def wait_for_running(jid, timeout_s=120):
+ deadline = time.time() + timeout_s
+ while time.time() < deadline:
+ vertices = _get(f"/jobs/{jid}")["vertices"]
+ if all(v["status"] == "RUNNING" for v in vertices):
+ return
+ time.sleep(1)
+ raise TimeoutError(f"job {jid} did not reach RUNNING")
+
+
+def compose(*args, mode=None, interval_ms=None):
+ env = dict(os.environ)
+ if mode:
+ env["LOAD_BALANCE_MODE"] = mode
+ if interval_ms is not None:
+ env["SLOT_REQUEST_MAX_INTERVAL"] = str(interval_ms)
+ subprocess.run(
+ ["docker", "compose", "-f", COMPOSE_FILE, *args],
+ check=True,
+ env=env,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ )
+
+
+def up(mode="NONE", interval_ms=20):
+ compose("up", "-d", "--force-recreate", mode=mode, interval_ms=interval_ms)
+ wait_for_cluster()
+
+
+def redeploy(mode, interval_ms=20, program_args=None):
+ """Apply a scheduling config: savepoint, recreate the cluster, restore the job."""
+ jid = running_job_id()
+ savepoint = stop_with_savepoint(jid) if jid else None
+ up(mode, interval_ms)
+ new_jid = submit(savepoint, program_args)
+ wait_for_running(new_jid)
+ return {"savepoint": savepoint, "job_id": new_jid}
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawTextHelpFormatter)
+ sub = parser.add_subparsers(dest="cmd", required=True)
+ sub.add_parser("status")
+ p_up = sub.add_parser("up")
+ p_up.add_argument("--mode", default="NONE")
+ p_up.add_argument("--interval", type=int, default=20)
+ p_submit = sub.add_parser("submit")
+ p_submit.add_argument("--savepoint")
+ p_submit.add_argument("job_args", nargs=argparse.REMAINDER)
+ p_redeploy = sub.add_parser("redeploy")
+ p_redeploy.add_argument("--mode", required=True)
+ p_redeploy.add_argument("--interval", type=int, default=20)
+ p_redeploy.add_argument("job_args", nargs=argparse.REMAINDER)
+ sub.add_parser("down")
+ args = parser.parse_args()
+
+ if args.cmd == "status":
+ result = status()
+ elif args.cmd == "up":
+ up(args.mode, args.interval)
+ result = cluster_config()
+ elif args.cmd == "submit":
+ jid = submit(args.savepoint, args.job_args)
+ wait_for_running(jid)
+ result = {"job_id": jid}
+ elif args.cmd == "redeploy":
+ result = redeploy(args.mode, args.interval, args.job_args)
+ else:
+ compose("down")
+ result = {"down": True}
+ json.dump(result, sys.stdout, indent=2)
+ print()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/flink-stateful-ops-agent-demo/target-jobs/clickstream-enrichment/pom.xml b/flink-stateful-ops-agent-demo/target-jobs/clickstream-enrichment/pom.xml
new file mode 100644
index 0000000..68a7a24
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/target-jobs/clickstream-enrichment/pom.xml
@@ -0,0 +1,74 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.flink.agents.demos</groupId>
+ <artifactId>clickstream-enrichment</artifactId>
+ <version>1.0</version>
+ <packaging>jar</packaging>
+
+ <properties>
+ <flink.version>2.2.1</flink.version>
+ <maven.compiler.release>11</maven.compiler.release>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+
+ <dependencies>
+ <!-- All provided by the Flink distribution (flink-dist bundles datagen). -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-datagen</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <finalName>clickstream-enrichment</finalName>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.11.0</version>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <version>3.3.0</version>
+ <configuration>
+ <archive>
+ <manifest>
+ <mainClass>org.apache.flink.agents.demos.clickstream.ClickstreamEnrichment</mainClass>
+ </manifest>
+ </archive>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/flink-stateful-ops-agent-demo/target-jobs/clickstream-enrichment/src/main/java/org/apache/flink/agents/demos/clickstream/ClickstreamEnrichment.java b/flink-stateful-ops-agent-demo/target-jobs/clickstream-enrichment/src/main/java/org/apache/flink/agents/demos/clickstream/ClickstreamEnrichment.java
new file mode 100644
index 0000000..9079b8d
--- /dev/null
+++ b/flink-stateful-ops-agent-demo/target-jobs/clickstream-enrichment/src/main/java/org/apache/flink/agents/demos/clickstream/ClickstreamEnrichment.java
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.demos.clickstream;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
+import org.apache.flink.connector.datagen.source.DataGeneratorSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.util.ParameterTool;
+
+/**
+ * Target job for the stateful ops demo: the job shape from the Flink "Balanced Tasks Scheduling"
+ * docs. Vertex A (source + Enrich) runs at parallelism 6 and vertex B (SessionScore + sink) at 3,
+ * in one slot sharing group. Both are CPU-bound, so the TaskManager that hosts the most tasks
+ * becomes the bottleneck under the default scheduling strategy.
+ *
+ * <p>Arguments: --rate (records/s for the whole job, 0 = unbounded), --work-a, --work-b
+ * (CPU iterations per record), --parallelism-a, --parallelism-b.
+ */
+public class ClickstreamEnrichment {
+
+ public static void main(String[] args) throws Exception {
+ ParameterTool params = ParameterTool.fromArgs(args);
+ double rate = params.getDouble("rate", 0);
+ int workA = params.getInt("work-a", 20_000);
+ int workB = params.getInt("work-b", 40_000);
+ int parallelismA = params.getInt("parallelism-a", 6);
+ int parallelismB = params.getInt("parallelism-b", 3);
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ DataGeneratorSource<Click> source =
+ new DataGeneratorSource<>(
+ index -> new Click(index % 100_000, index),
+ Long.MAX_VALUE,
+ rate > 0 ? RateLimiterStrategy.perSecond(rate) : RateLimiterStrategy.noOp(),
+ TypeInformation.of(Click.class));
+
+ env.fromSource(source, WatermarkStrategy.noWatermarks(), "Clicks")
+ .uid("clicks")
+ .setParallelism(parallelismA)
+ .map(new Enrich(workA))
+ .name("Enrich")
+ .uid("enrich")
+ .setParallelism(parallelismA)
+ .keyBy(click -> click.userId)
+ .map(new SessionScore(workB))
+ .name("SessionScore")
+ .uid("session-score")
+ .setParallelism(parallelismB)
+ .sinkTo(new DiscardingSink<>())
+ .name("Sink")
+ .uid("sink")
+ .setParallelism(parallelismB);
+
+ env.execute("ClickstreamEnrichment");
+ }
+
+ /** One click event. */
+ public static class Click {
+ public long userId;
+ public long seq;
+ public long checksum;
+
+ public Click() {}
+
+ public Click(long userId, long seq) {
+ this.userId = userId;
+ this.seq = seq;
+ }
+ }
+
+ /** Stand-in for CPU-heavy enrichment work (parsing, lookups, feature extraction). */
+ static long burn(long seed, int iterations) {
+ long x = seed | 1;
+ for (int i = 0; i < iterations; i++) {
+ x ^= x << 13;
+ x ^= x >>> 7;
+ x ^= x << 17;
+ }
+ return x;
+ }
+
+ static class Enrich extends RichMapFunction<Click, Click> {
+ private final int work;
+
+ Enrich(int work) {
+ this.work = work;
+ }
+
+ @Override
+ public Click map(Click click) {
+ click.checksum = burn(click.seq, work);
+ return click;
+ }
+ }
+
+ /** Keyed and stateful, so the savepoint carries real state across redeploys. */
+ static class SessionScore extends RichMapFunction<Click, Long> {
+ private final int work;
+ private transient ValueState<Long> clicksPerUser;
+
+ SessionScore(int work) {
+ this.work = work;
+ }
+
+ @Override
+ public void open(OpenContext openContext) {
+ clicksPerUser =
+ getRuntimeContext()
+ .getState(new ValueStateDescriptor<>("clicks-per-user", Long.class));
+ }
+
+ @Override
+ public Long map(Click click) throws Exception {
+ Long count = clicksPerUser.value();
+ long next = count == null ? 1 : count + 1;
+ clicksPerUser.update(next);
+ return burn(click.checksum ^ next, work);
+ }
+ }
+}