[knechtionscoding]

What type of PR is this?

/kind api

What this PR does / why we need it:

Gravity starts over between Slack messages. Each message spawns a fresh Task
that runs the agent with no conversation history, so iterating in a thread does
not work the way iterating in a Claude Code conversation does — you re-explain
context every turn.

Session and SessionSpawner already cover a conversation's lifecycle, but
nothing could drive a turn from outside a browser, and nothing connected Slack
to any of it. This PR adds both halves.

A turn driver — internal/sessionturn. There is no request/response API for
a Session turn: the runtime listens on a unix socket inside the Session Pod, and
the console server reaches it by pods/exec-ing /kelos/bin/kelos-session-runtime.
This package speaks that protocol over the same bridge. It subscribes, submits
one message request with a request ID, learns its turn ID from the echoed user
message, follows only that turn's events, and returns the assistant reply on
turn.completed. It handles a turn already in flight (the runtime queues behind
it), a prompt merged into an already-queued turn, a stream that drops mid-turn,
and a 60-minute bound on one turn. WaitReady resolves a Session to its Pod,
requesting a resume when the idle policy suspended it.

The Slack ingress. SessionSpawner.spec.when gains a slack source,
mutually exclusive with githubWebhook, reusing the channel, trigger, exclusion
and bot-message rules TaskSpawner's Slack source already has.
sessionTemplate.initialPrompt and sessionTemplate.suspend are rejected for a
Slack source: every matching thread message already runs as its own turn, so an
initial prompt would only delay the message that asked for something, and a
suspended Session cannot answer the thread that created it.

Every prompt the bridge submits begins with a fixed envelope naming the author.
That is load-bearing, not cosmetic: the runtime parses a prompt for its own
commands before the agent sees it, and text beginning with ! runs as a shell
command in the Session Pod with the Session's credentials.

The Slack server maps channel + thread_ts to a deterministic Session name
(hashed, since a Slack timestamp contains a dot), creates the Session on the
thread's first matching message and reuses it for every later matching message
in that thread — an unmatched reply is ignored and never reaches the Session. Each
thread gets its own serial queue, so a message arriving mid-turn waits rather
than opening a second stream into the same Session; up to eight may wait before
the bot says so in the thread.

What it looks like in Slack. The bot posts a :hourglass: Working on your request... reply into the thread, then edits that same message in place with
the answer once the turn finishes. A reply too long for one Slack message
continues in further thread replies. A failed turn, an interrupted turn, or a
Session that never becomes ready reports the reason in the same place rather
than leaving the thread waiting.

Which issue(s) this PR is related to:

N/A — tracked in GRAV-1028.

Special notes for your reviewer:

The ticket proposed this as a stack of two PRs. It is one here because the turn
driver alone cannot be exercised in Slack, and the point of shipping this is to
see what a threaded conversation actually feels like.

Decisions worth a second opinion:

• Where the driver runs. In the Slack server process, not a separate
  controller. A turn does not survive a Slack server restart; the Session and
  its history do, so the next message in the thread picks up where it left off.
• Concurrency within a thread. Queued, in arrival order, one at a time.
  The runtime also merges a prompt into an already-queued turn, which the driver
  follows rather than fighting.
• Lifecycle. Every matching thread gets a Pod and, with a
  volumeClaimTemplate, a PVC. idlePolicy is the lever; the example sets
  suspendAfterSeconds: 1800 and deleteAfterSeconds: 86400. Worth picking
  real numbers before the first deploy rather than after the cluster fills.
• Scope. spec.when.slack.channels keeps the first deploy to one channel.
• A TaskSpawner and a SessionSpawner that both match a message each act on it,
  producing two replies. Documented; not enforced.

RBAC: the Slack server now needs create on pods/exec, get/list/watch
on sessionspawners, and create/get/list/watch/patch on sessions.
The chart grants these.

No UI changes, so no screenshots. make verify, make test and
make test-integration all pass locally (166 + 12 integration specs).

Does this PR introduce a user-facing change?

SessionSpawner now accepts a `spec.when.slack` source. A matching Slack thread gets its own persistent Session: the first matching message creates it and every later matching message in that thread runs as another turn in the same conversation, so the agent keeps what it has already been told. A reply that does not match — by default, one that does not @-mention the bot — is ignored and never reaches the Session. The reply is posted into the thread and edited in place as the turn finishes. `spec.sessionTemplate.initialPrompt` and `spec.sessionTemplate.suspend` are rejected for a Slack source.

🤖 Generated with Claude Code

---

Summary by cubic

Gives each Slack thread a persistent Session so matching messages in a thread run as turns of one conversation, instead of each message spawning a fresh Task that starts with no history.

Slack-driven Sessions

• Adds internal/sessionturn, which drives one turn by speaking the Session runtime's socket protocol over the pods/exec bridge the console server uses.
• Adds spec.when.slack to SessionSpawner, mutually exclusive with githubWebhook; sessionTemplate.initialPrompt and sessionTemplate.suspend are rejected for a Slack source.
• The Slack server maps channel + thread_ts to a deterministic Session name, creates the Session on the thread's first matching message, and reuses it for later ones. Messages arriving mid-turn wait per-thread; up to eight may queue before the bot reports the drop.
• Every prompt starts with a fixed author envelope — the runtime parses !-prefixed text as shell commands — so no user-controlled Slack text can reach the runtime first.
• The bot posts a placeholder reply and edits it in place with the result; failed or interrupted turns report the reason there.

RBAC

• The Slack server now needs create on pods/exec, get/list/watch on sessionspawners, and create/get/list/watch/patch on sessions; the chart grants these.

^{Written for commit 2932424. Summary will update on new commits.}