fix(compose): the worker's store init now runs on every up, after the core

docker-compose.yml

@@ -213,7 +213,7 @@ services:
       start_period: 40s
 
   # PostgreSQL delivery store for the Go Worker. Profile-gated on `worker`
-  # for the same reason worker-store-init is: a stock deployment runs on
+  # because it is only ever an opt-in: a stock deployment runs on
   # SQLite and must not gain a database it never asked for. An operator opts
   # the durable queue onto PostgreSQL by setting ATF_DATABASE_URL on the
   # worker (and the Core, which owns the schema bootstrap).
@@ -244,15 +244,25 @@ services:
       retries: 20
     profiles: ["worker"]
 
-  # One-time preparation of the shared delivery store for the non-root
-  # worker. The Core creates atf.db as root; the worker (uid 100) must be
-  # able to write its claim updates to the SAME file. This runs only under
-  # the `worker` profile (i.e. when the operator opts into the Go Worker),
-  # so a stock deployment never sees it.
+  # Preparation of the shared delivery store for the non-root worker.
+  # The Core creates atf.db as root; the worker (uid 100) must be able to
+  # write its claim updates to the SAME file. This runs on EVERY `compose
+  # up` — the worker container itself is not profile-gated (it always runs
+  # to report health; only execution is gated by ATF_WORKER_ENABLED), so a
+  # stock deployment needs this too. It waits for the core to be healthy
+  # because on a fresh volume atf.db does not exist until the core creates
+  # it: an init that ran earlier would chgrp an empty directory and leave
+  # the newly created file root-owned — the exact crash loop
+  # ("probe write: attempt to write a readonly database") that failed the
+  # v2.5.0 clean-install gate. Idempotent: a second run is a no-op, so
+  # re-running after a Core restart is safe.
   worker-store-init:
     image: alpine:3.22
     container_name: atf-worker-store-init
     restart: "no"
+    depends_on:
+      core:
+        condition: service_healthy
     volumes:
       - atf-data:/data
     # chgrp to the worker's gid (101 in the alpine image, from `addgroup -S`)
@@ -261,7 +271,6 @@ services:
     command: >
       sh -c "chgrp -R 101 /data && chmod -R g+w /data || true;
              echo 'shared store prepared for the atf group'"
-    profiles: ["worker"]
 
   # Go execution engine (business execution plane). OFF by default:
   # ATF_WORKER_ENABLED must be set (AND ATF_CORE_RPC_TOKEN must be non-empty)
@@ -344,6 +353,11 @@ services:
     depends_on:
       core:
         condition: service_healthy
+      # The store init must have chgrp'd/chmod'd atf.db before this
+      # container's probe write — without the wait the worker fail-closes
+      # on a root-owned database and restart-loops forever.
+      worker-store-init:
+        condition: service_completed_successfully
       # `required: false` is what makes the profile gate work: when the
       # `worker` profile is off, postgres does not exist and compose skips
       # this dependency instead of failing; when the profile is on, the

worker/cmd/worker/main.go

@@ -53,6 +53,7 @@ import (
 	"atf/worker/internal/telegraph"
 	"atf/worker/internal/telegraph/fakegateway"
 	"atf/worker/internal/telegraph/grpcclient"
+	"atf/worker/internal/telegraph/unconfigured"
 )
 
 // ContractVersion is the Telegram Gateway contract this build speaks. It
@@ -174,10 +175,25 @@ func run() (runErr error) {
 	default:
 		client, err := grpcclient.New(cfg.TelegramRuntimeAddr, cfg.CoreRPCToken, workerID(), 60*time.Second)
 		if err != nil {
-			return err
+			if !cfg.WorkerEnabled {
+				// The documented contract: an empty token refuses to
+				// start only an ENABLED worker. A disabled one exists to
+				// report health — the strangler-fig guarantee that the
+				// Python Core owns execution until the operator moves the
+				// boundary — so it boots with the health-only gateway
+				// instead of restart-looping with a dark /ops endpoint
+				// (the crash loop the v2.5.0 clean-install gate caught).
+				log.Warn("Telegram runtime client unconfigured — running health-only; enable the worker only after configuring it",
+					"error", err,
+					"telegram_runtime", cfg.TelegramRuntimeAddr)
+				gateway = unconfigured.New(err.Error())
+			} else {
+				return err
+			}
+		} else {
+			defer client.Close()
+			gateway = client
 		}
-		defer client.Close()
-		gateway = client
 	}
 
 	rules, err := openRuleStore(context.Background(), cfg)

worker/internal/telegraph/unconfigured/unconfigured.go

@@ -0,0 +1,168 @@
+// Package unconfigured is the health-only telegraph.Gateway of a worker
+// that cannot reach the Telegram runtime.
+//
+// The worker container always runs — even before the operator opts into Go
+// execution — so it can report health. A worker in that state has no usable
+// Telegram transport: ATF_CORE_RPC_TOKEN is empty, so the real gRPC client
+// refuses to construct ("refusing an unauthenticated client"). Rather than
+// refusing to START — which would leave the ops endpoint dark, the API's
+// health payload degraded, and the container restart-looping (the crash
+// loop the v2.5.0 clean-install gate caught) — the process boots with this
+// gateway: every operation fails closed with a failure that names the
+// missing configuration and the operation that was never attempted.
+//
+// The delivery engine never reaches it while ATF_WORKER_ENABLED is off.
+// The moment the operator enables the worker, an unconfigured token is
+// fatal at startup again — fail closed, on purpose.
+//
+// Like the fake gateway, this is never a substitute for a real delivery:
+// nothing sent through it happened.
+package unconfigured
+
+import (
+	"context"
+
+	"atf/worker/internal/telegraph"
+)
+
+// Code is the canonical failure category for an operation that was refused
+// because the worker has no Telegram transport configured. Not in
+// DeliveryFailure.IsTransient: a missing configuration does not heal by
+// retrying.
+const Code = "TRANSPORT_UNCONFIGURED"
+
+// Gateway is the telegraph.Gateway of a health-only worker.
+type Gateway struct {
+	// reason is the startup error the real client refused with, surfaced
+	// verbatim so the operator reads the actual cause, not a paraphrase.
+	reason string
+}
+
+// New returns a Gateway whose every operation fails with reason.
+func New(reason string) *Gateway {
+	return &Gateway{reason: reason}
+}
+
+// refusal builds the failure every operation returns. The operation name is
+// spelled out so a log line or Log Center row answers "what did NOT happen"
+// without reading the stack.
+func (g *Gateway) refusal(op string) *telegraph.DeliveryFailure {
+	return &telegraph.DeliveryFailure{
+		Code: Code,
+		Message: "worker runs health-only: Telegram transport unconfigured (" +
+			g.reason + ") — " + op + " never attempted",
+	}
+}
+
+// IsSessionReady reports the transport as never ready: a connecting session
+// is not ready either, so callers defer rather than fail the job.
+func (g *Gateway) IsSessionReady(_ context.Context, _ string) (bool, string, error) {
+	return false, "transport unconfigured (" + g.reason + ")", nil
+}
+
+// --- reads ------------------------------------------------------------- //
+
+// GetMessage is refused: see the package comment.
+func (g *Gateway) GetMessage(_ context.Context, _ string, _, _ int64) (*telegraph.MessageSnapshot, *telegraph.DeliveryFailure) {
+	return nil, g.refusal("GetMessage")
+}
+
+// GetAlbumMessageIds is refused: see the package comment.
+func (g *Gateway) GetAlbumMessageIds(_ context.Context, _ string, _, _ int64) ([]int64, *telegraph.DeliveryFailure) {
+	return nil, g.refusal("GetAlbumMessageIds")
+}
+
+// GetFileExtension is refused: see the package comment.
+func (g *Gateway) GetFileExtension(_ context.Context, _ string, _, _ int64) (string, *telegraph.DeliveryFailure) {
+	return "", g.refusal("GetFileExtension")
+}
+
+// --- media staging ------------------------------------------------------ //
+
+// DownloadMedia is refused: see the package comment.
+func (g *Gateway) DownloadMedia(_ context.Context, _ string, _, _ int64, _ telegraph.MediaHandle) (int64, *telegraph.DeliveryFailure) {
+	return 0, g.refusal("DownloadMedia")
+}
+
+// ReloadMedia is refused: see the package comment.
+func (g *Gateway) ReloadMedia(_ context.Context, _ string, _, _ int64, _ telegraph.MediaHandle, _ string) (int64, *telegraph.DeliveryFailure) {
+	return 0, g.refusal("ReloadMedia")
+}
+
+// PrepareMediaUpload is refused: see the package comment.
+func (g *Gateway) PrepareMediaUpload(_ context.Context, _ string, _ telegraph.MediaHandle, _, _ int64, _ string) (string, *telegraph.DeliveryFailure) {
+	return "", g.refusal("PrepareMediaUpload")
+}
+
+// --- sends (the Telegram side effects) ---------------------------------- //
+
+// ForwardMessages is refused: see the package comment.
+func (g *Gateway) ForwardMessages(_ context.Context, _ string, _, _ int64, _ []int64, _ int64) ([]int64, *telegraph.DeliveryFailure) {
+	return nil, g.refusal("ForwardMessages")
+}
+
+// CopyMessage is refused: see the package comment.
+func (g *Gateway) CopyMessage(_ context.Context, _ string, _, _, _ int64, _ string, _ int64, _ bool) (int64, *telegraph.DeliveryFailure) {
+	return 0, g.refusal("CopyMessage")
+}
+
+// CopyMessages is refused: see the package comment.
+func (g *Gateway) CopyMessages(_ context.Context, _ string, _, _ int64, _ []int64, _ string, _ int64, _ bool) ([]int64, *telegraph.DeliveryFailure) {
+	return nil, g.refusal("CopyMessages")
+}
+
+// CopyMediaGroup is refused: see the package comment.
+func (g *Gateway) CopyMediaGroup(_ context.Context, _ string, _, _ int64, _ int64, _ string, _ int64) ([]int64, *telegraph.DeliveryFailure) {
+	return nil, g.refusal("CopyMediaGroup")
+}
+
+// SendMessage is refused: see the package comment.
+func (g *Gateway) SendMessage(_ context.Context, _ string, _ int64, _ string, _ int64) (int64, *telegraph.DeliveryFailure) {
+	return 0, g.refusal("SendMessage")
+}
+
+// SendMedia is refused: see the package comment.
+func (g *Gateway) SendMedia(_ context.Context, _ string, _ int64, _, _, _ string, _ int64) (int64, *telegraph.DeliveryFailure) {
+	return 0, g.refusal("SendMedia")
+}
+
+// --- bot transport ------------------------------------------------------ //
+
+// BotAvailability is refused: see the package comment.
+func (g *Gateway) BotAvailability(_ context.Context, _ telegraph.BotTarget) (bool, string, *telegraph.DeliveryFailure) {
+	return false, "", g.refusal("BotAvailability")
+}
+
+// BotSendText is refused: see the package comment.
+func (g *Gateway) BotSendText(_ context.Context, _ telegraph.BotTarget, _ int64, _ string, _ int64, _ []string) (int64, *telegraph.DeliveryFailure) {
+	return 0, g.refusal("BotSendText")
+}
+
+// BotCopyMessage is refused: see the package comment.
+func (g *Gateway) BotCopyMessage(_ context.Context, _ telegraph.BotTarget, _, _, _ int64, _ string, _ int64, _ []string) (int64, *telegraph.DeliveryFailure) {
+	return 0, g.refusal("BotCopyMessage")
+}
+
+// BotSendMedia is refused: see the package comment.
+func (g *Gateway) BotSendMedia(_ context.Context, _ telegraph.BotTarget, _ int64, _, _, _ string, _ int64, _ []string) (int64, *telegraph.DeliveryFailure) {
+	return 0, g.refusal("BotSendMedia")
+}
+
+// BotDeleteMessages is refused: see the package comment.
+func (g *Gateway) BotDeleteMessages(_ context.Context, _ telegraph.BotTarget, _ int64, _ []int64) (bool, *telegraph.DeliveryFailure) {
+	return false, g.refusal("BotDeleteMessages")
+}
+
+// --- deletion (auto-delete scheduler) ----------------------------------- //
+
+// DeleteMessages is refused: see the package comment.
+func (g *Gateway) DeleteMessages(_ context.Context, _ telegraph.DeleteTransport, _, _ string, _, _ int64, _ []int64) (bool, *telegraph.DeliveryFailure) {
+	return false, g.refusal("DeleteMessages")
+}
+
+// --- branding stage 1 ---------------------------------------------------- //
+
+// BrandingStage1 is refused: see the package comment.
+func (g *Gateway) BrandingStage1(_ context.Context, _ telegraph.BrandingPost) ([]int64, *telegraph.DeliveryFailure) {
+	return nil, g.refusal("BrandingStage1")
+}

worker/internal/telegraph/unconfigured/unconfigured_test.go

@@ -0,0 +1,101 @@
+package unconfigured
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"atf/worker/internal/telegraph"
+)
+
+// A compile-time shape check: the health-only gateway is a telegraph.Gateway,
+// so it can drop into cmd/worker's disabled path.
+var _ telegraph.Gateway = (*Gateway)(nil)
+
+// The health-only gateway must fail EVERY operation with the canonical
+// non-transient code and a message that names the missing configuration —
+// never a nil failure (which reads as "the operation happened") and never a
+// panic (which would take the ops endpoint down with it).
+func TestEveryOperationFailsClosed(t *testing.T) {
+	g := New("internal RPC token is empty: refusing an unauthenticated client")
+	ctx := context.Background()
+	target := telegraph.BotTarget{TokenID: "t1", OwnerUserID: 7}
+
+	if ok, reason, err := g.IsSessionReady(ctx, "s1"); ok || err != nil || !strings.Contains(reason, "unconfigured") {
+		t.Fatalf("IsSessionReady = (%v, %q, %v), want (false, unconfigured reason, nil)", ok, reason, err)
+	}
+
+	_, f := g.GetMessage(ctx, "s1", 1, 2)
+	check(t, "GetMessage", f, g.reason)
+	_, f = g.GetAlbumMessageIds(ctx, "s1", 1, 2)
+	check(t, "GetAlbumMessageIds", f, g.reason)
+	_, f = g.GetFileExtension(ctx, "s1", 1, 2)
+	check(t, "GetFileExtension", f, g.reason)
+	_, f = g.DownloadMedia(ctx, "s1", 1, 2, telegraph.MediaHandle{})
+	check(t, "DownloadMedia", f, g.reason)
+	_, f = g.ReloadMedia(ctx, "s1", 1, 2, telegraph.MediaHandle{}, "photo")
+	check(t, "ReloadMedia", f, g.reason)
+	_, f = g.PrepareMediaUpload(ctx, "s1", telegraph.MediaHandle{}, 1, 2, "photo")
+	check(t, "PrepareMediaUpload", f, g.reason)
+	_, f = g.ForwardMessages(ctx, "s1", 1, 2, []int64{3}, 0)
+	check(t, "ForwardMessages", f, g.reason)
+	_, f = g.CopyMessage(ctx, "s1", 1, 2, 3, "", 0, false)
+	check(t, "CopyMessage", f, g.reason)
+	_, f = g.CopyMessages(ctx, "s1", 1, 2, []int64{3}, "", 0, false)
+	check(t, "CopyMessages", f, g.reason)
+	_, f = g.CopyMediaGroup(ctx, "s1", 1, 2, 3, "", 0)
+	check(t, "CopyMediaGroup", f, g.reason)
+	_, f = g.SendMessage(ctx, "s1", 1, "text", 0)
+	check(t, "SendMessage", f, g.reason)
+	_, f = g.SendMedia(ctx, "s1", 1, "tok", "cap", "photo", 0)
+	check(t, "SendMedia", f, g.reason)
+	_, f = g.BotSendText(ctx, target, 1, "text", 0, nil)
+	check(t, "BotSendText", f, g.reason)
+	_, f = g.BotCopyMessage(ctx, target, 1, 2, 3, "", 0, nil)
+	check(t, "BotCopyMessage", f, g.reason)
+	_, f = g.BotSendMedia(ctx, target, 1, "tok", "cap", "photo", 0, nil)
+	check(t, "BotSendMedia", f, g.reason)
+	_, f = g.BrandingStage1(ctx, telegraph.BrandingPost{})
+	check(t, "BrandingStage1", f, g.reason)
+
+	ok, username, f := g.BotAvailability(ctx, target)
+	if ok || username != "" {
+		t.Fatalf("BotAvailability = (%v, %q), want (false, \"\")", ok, username)
+	}
+	check(t, "BotAvailability", f, g.reason)
+
+	ok, f = g.BotDeleteMessages(ctx, target, 1, []int64{2})
+	if ok {
+		t.Fatalf("BotDeleteMessages = %v, want false", ok)
+	}
+	check(t, "BotDeleteMessages", f, g.reason)
+
+	ok, f = g.DeleteMessages(ctx, telegraph.TransportUser, "s1", "b1", 7, 1, []int64{2})
+	if ok {
+		t.Fatalf("DeleteMessages = %v, want false", ok)
+	}
+	check(t, "DeleteMessages", f, g.reason)
+}
+
+// check asserts one refusal.
+func check(t *testing.T, op string, f *telegraph.DeliveryFailure, reason string) {
+	t.Helper()
+	if f == nil {
+		t.Fatalf("%s returned nil failure — a nil reads as \"the operation happened\"", op)
+	}
+	if f.Code != Code {
+		t.Errorf("%s failure code = %q, want %q (message: %s)", op, f.Code, Code, f.Message)
+	}
+	if f.IsTransient() {
+		t.Errorf("%s: %q must not be transient — a missing configuration does not heal by retrying", op, f.Code)
+	}
+	if !strings.Contains(f.Message, "health-only") {
+		t.Errorf("%s refusal must name the health-only state, got: %s", op, f.Message)
+	}
+	if !strings.Contains(f.Message, reason) {
+		t.Errorf("%s refusal must carry the startup reason verbatim, got: %s", op, f.Message)
+	}
+	if !strings.Contains(f.Message, op) {
+		t.Errorf("%s refusal must name the operation that was never attempted, got: %s", op, f.Message)
+	}
+}