refactor(backend): drop backend-driven Fabric broadcast — agent-driven model
The backend no longer broadcasts topic lifecycle events to Fabric. The
new model (per design discussion 2026-05-23 evening):
- Proposing agent posts a single recruitment fabric-send-message
immediately after creating a topic (carries topic_id + signup
window + debate window + title).
- Downstream agents that decide to participate book a HF on_call
slot covering the debate window via `hf calendar schedule on_call
<time> <duration> --job DEBATE-<topic_id>`.
- HF wakes the agent naturally at slot start; the wake payload
carries event_data with the DEBATE-<topic_id> code so the agent
knows why it was woken.
- The backend stays a pure data + state-machine service and doesn't
know about Fabric.
Code removed:
- internal/fabric/announce.go (entire file + empty dir)
- ticker.go: broadcastLifecycle + broadcastAnnouncement + topicTarget
helpers; announcer field on Ticker; announce field/arg on NewTicker
- models/topic.go: AnnounceGuildBaseURL + AnnounceChannelID fields
- store/topic_store.go: same fields on CreateTopicInput + INSERT
- handlers/topics.go: same fields on createTopicBody + validation +
parameter passing to store
- handlers/verdict.go: announcer field + lifecycle broadcast on
verdict submit
- config/config.go: FabricSystemAPIKey field + DIALECTIC_FABRIC_SYSTEM_API_KEY
env read
- main.go + routes.go: announcer wiring
Database:
- migrations/003_drop_topic_announce_target.sql drops the two columns
added by migration 002. Counterpart commit on the deployment side
needs DIALECTIC_FABRIC_SYSTEM_API_KEY env removed from
docker-compose.yml; harmless if left as the backend no longer
reads it.
Pairs with:
- Dialectic.OpenclawPlugin: rip announce_* params from
dialectic_propose_topic (next commit)
- Fabric.Backend.Center: rip serviceEndpoint field + cli
- Fabric.Backend.Guild: rip system-key bypass on ApiKeyGuard and
announce-only-system limit on messaging.controller
- ClawSkills: rewrite participate-debate + analyze-intel step 4 +
delete rotate-fabric-system-key workflow
This commit is contained in:
@@ -66,24 +66,14 @@ type Config struct {
|
|||||||
OIDCIssuer string
|
OIDCIssuer string
|
||||||
OIDCClientID string
|
OIDCClientID string
|
||||||
|
|
||||||
// Fabric announce coupling.
|
// (Removed Aug 2026: all Fabric coupling — FabricSystemAPIKey,
|
||||||
//
|
// FabricGuildBaseURL, FabricAnnounceChannelID, FabricBotBearerToken.
|
||||||
// As of Phase 3.5: only the system api key stays in env. Guild base
|
// Backend no longer broadcasts lifecycle events to Fabric. The
|
||||||
// URL + announce channel ID are PER-TOPIC, supplied by the proposing
|
// proposing agent posts a single recruitment fabric-send-message
|
||||||
// agent at create time and stored on the topic row. Different topics
|
// after creating a topic; downstream agents book HF on_call slots
|
||||||
// can broadcast to different guilds/channels from the same backend.
|
// covering the debate window via `hf calendar schedule` and HF
|
||||||
//
|
// wakes them naturally. The backend stays a pure data + state-
|
||||||
// FabricSystemAPIKey: header value for x-fabric-system-key when
|
// machine service and doesn't know about Fabric.)
|
||||||
// POSTing to announce channels. Must match Fabric.Backend.Guild's
|
|
||||||
// FABRIC_BACKEND_GUILD_SYSTEM_API_KEY env. Empty → announcer becomes
|
|
||||||
// a no-op (logs intent, skips post — useful for dev / opted-out
|
|
||||||
// environments).
|
|
||||||
FabricSystemAPIKey string
|
|
||||||
|
|
||||||
// (removed Phase 3.5: FabricGuildBaseURL, FabricAnnounceChannelID,
|
|
||||||
// FabricBotBearerToken — guild/channel moved to per-topic config;
|
|
||||||
// bot bearer obsolete since Guild's ApiKeyGuard now accepts system
|
|
||||||
// key alone for announce posts.)
|
|
||||||
|
|
||||||
// Orchestrator tick interval. 0 / unset → default 15s.
|
// Orchestrator tick interval. 0 / unset → default 15s.
|
||||||
OrchestratorTickInterval time.Duration
|
OrchestratorTickInterval time.Duration
|
||||||
@@ -105,7 +95,6 @@ func LoadFromEnv() (*Config, error) {
|
|||||||
DialecticAdminAPIKey: os.Getenv("DIALECTIC_ADMIN_API_KEY"),
|
DialecticAdminAPIKey: os.Getenv("DIALECTIC_ADMIN_API_KEY"),
|
||||||
OIDCIssuer: os.Getenv("OIDC_ISSUER"),
|
OIDCIssuer: os.Getenv("OIDC_ISSUER"),
|
||||||
OIDCClientID: os.Getenv("OIDC_CLIENT_ID"),
|
OIDCClientID: os.Getenv("OIDC_CLIENT_ID"),
|
||||||
FabricSystemAPIKey: os.Getenv("DIALECTIC_FABRIC_SYSTEM_API_KEY"),
|
|
||||||
}
|
}
|
||||||
if d := os.Getenv("ORCHESTRATOR_TICK_INTERVAL"); d != "" {
|
if d := os.Getenv("ORCHESTRATOR_TICK_INTERVAL"); d != "" {
|
||||||
if parsed, err := time.ParseDuration(d); err == nil {
|
if parsed, err := time.ParseDuration(d); err == nil {
|
||||||
|
|||||||
10
internal/db/migrations/003_drop_topic_announce_target.sql
Normal file
10
internal/db/migrations/003_drop_topic_announce_target.sql
Normal file
@@ -0,0 +1,10 @@
|
|||||||
|
-- Drop per-topic announce target columns. The backend no longer
|
||||||
|
-- broadcasts lifecycle events to Fabric; the proposing agent posts a
|
||||||
|
-- single recruitment fabric-send-message after topic creation, and
|
||||||
|
-- downstream agents book HF on_call slots covering the debate window
|
||||||
|
-- via `hf calendar schedule` so HF wakes them naturally.
|
||||||
|
--
|
||||||
|
-- Counterpart of 002_topic_announce_target.sql (now obsolete).
|
||||||
|
ALTER TABLE topics
|
||||||
|
DROP COLUMN announce_guild_base_url,
|
||||||
|
DROP COLUMN announce_channel_id;
|
||||||
@@ -1,131 +0,0 @@
|
|||||||
// Package fabric provides an HTTP client for posting system messages
|
|
||||||
// to a Fabric Guild's announce-type channel.
|
|
||||||
//
|
|
||||||
// As of Phase 3.5 the target (guildBaseUrl + channelID) is per-call
|
|
||||||
// rather than backend-env: each topic stores its own announce target,
|
|
||||||
// chosen by the proposing agent at create time. Only the system api
|
|
||||||
// key stays in backend env (DIALECTIC_FABRIC_SYSTEM_API_KEY) since it
|
|
||||||
// authorizes ANY announce POST regardless of channel.
|
|
||||||
//
|
|
||||||
// FABRIC_BOT_BEARER_TOKEN is gone — the upgraded Fabric.Backend.Guild
|
|
||||||
// ApiKeyGuard accepts x-fabric-system-key alone (no per-user Bearer
|
|
||||||
// required) for announce-channel posts.
|
|
||||||
package fabric
|
|
||||||
|
|
||||||
import (
|
|
||||||
"bytes"
|
|
||||||
"context"
|
|
||||||
"encoding/json"
|
|
||||||
"fmt"
|
|
||||||
"io"
|
|
||||||
"log"
|
|
||||||
"net/http"
|
|
||||||
"time"
|
|
||||||
)
|
|
||||||
|
|
||||||
type Announcer struct {
|
|
||||||
systemAPIKey string
|
|
||||||
requestTimeout time.Duration
|
|
||||||
client *http.Client
|
|
||||||
}
|
|
||||||
|
|
||||||
func NewAnnouncer(systemAPIKey string) *Announcer {
|
|
||||||
timeout := 5 * time.Second
|
|
||||||
return &Announcer{
|
|
||||||
systemAPIKey: systemAPIKey,
|
|
||||||
requestTimeout: timeout,
|
|
||||||
client: &http.Client{Timeout: timeout},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Enabled returns true iff the system api key is configured (without
|
|
||||||
// it, no announce POST can authenticate).
|
|
||||||
func (a *Announcer) Enabled() bool { return a.systemAPIKey != "" }
|
|
||||||
|
|
||||||
// Target identifies one announce post destination. Both fields must
|
|
||||||
// be non-empty; either empty -> Post* returns a "skipped" no-op.
|
|
||||||
type Target struct {
|
|
||||||
GuildBaseURL string
|
|
||||||
ChannelID string
|
|
||||||
}
|
|
||||||
|
|
||||||
// PostTopicAnnouncement: first signup-open broadcast.
|
|
||||||
// PostLifecycleEvent: subsequent state transitions.
|
|
||||||
//
|
|
||||||
// Both best-effort: log + nil on transport / target-misconfigured
|
|
||||||
// errors so the orchestrator transition isn't blocked by a Fabric
|
|
||||||
// outage or an opted-out topic.
|
|
||||||
|
|
||||||
func (a *Announcer) PostTopicAnnouncement(ctx context.Context, t Target,
|
|
||||||
topicID, title, summary string,
|
|
||||||
signupOpen, signupClose, debateStart, debateEnd time.Time,
|
|
||||||
verdictSchemaID string) error {
|
|
||||||
|
|
||||||
body := formatAnnouncement(topicID, title, summary,
|
|
||||||
signupOpen, signupClose, debateStart, debateEnd, verdictSchemaID)
|
|
||||||
return a.post(ctx, t, topicID, "signup_open", body)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *Announcer) PostLifecycleEvent(ctx context.Context, t Target,
|
|
||||||
topicID, title, kind, summary string) error {
|
|
||||||
body := fmt.Sprintf("📣 **[%s]** %s [%s]\n%s", kind, title, topicID, summary)
|
|
||||||
return a.post(ctx, t, topicID, kind, body)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *Announcer) post(ctx context.Context, t Target, topicID, kind, content string) error {
|
|
||||||
if !a.Enabled() {
|
|
||||||
log.Printf("announce: skipped topic=%s kind=%s — DIALECTIC_FABRIC_SYSTEM_API_KEY unset", topicID, kind)
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
if t.GuildBaseURL == "" || t.ChannelID == "" {
|
|
||||||
log.Printf("announce: skipped topic=%s kind=%s — no announce target on topic (creator opted out)", topicID, kind)
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
body, _ := json.Marshal(map[string]any{"content": content})
|
|
||||||
url := fmt.Sprintf("%s/api/channels/%s/messages",
|
|
||||||
trimRightSlash(t.GuildBaseURL), t.ChannelID)
|
|
||||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
req.Header.Set("content-type", "application/json")
|
|
||||||
req.Header.Set("x-fabric-system-key", a.systemAPIKey)
|
|
||||||
|
|
||||||
resp, err := a.client.Do(req)
|
|
||||||
if err != nil {
|
|
||||||
log.Printf("announce: POST %s failed: %v", url, err)
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
defer resp.Body.Close()
|
|
||||||
if resp.StatusCode >= 300 {
|
|
||||||
b, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
|
|
||||||
log.Printf("announce: POST %s kind=%s -> %d body=%s", url, kind, resp.StatusCode, string(b))
|
|
||||||
return fmt.Errorf("announce post: status %d", resp.StatusCode)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func formatAnnouncement(id, title, summary string,
|
|
||||||
signupOpen, signupClose, debateStart, debateEnd time.Time, schema string) string {
|
|
||||||
const layout = "2006-01-02 15:04 UTC"
|
|
||||||
return fmt.Sprintf(
|
|
||||||
"🆕 **Debate signup open** [%s]\n\n**%s**\n%s\n\n"+
|
|
||||||
"• Signup: %s → %s\n"+
|
|
||||||
"• Debate: %s → %s\n"+
|
|
||||||
"• Verdict schema: `%s`\n\n"+
|
|
||||||
"To volunteer, use the `dialectic_signup` tool with this topic_id and your willing camp(s) — pro / con / judge. "+
|
|
||||||
"You must have an `on_call` slot covering the debate window.",
|
|
||||||
id, title, summary,
|
|
||||||
signupOpen.UTC().Format(layout), signupClose.UTC().Format(layout),
|
|
||||||
debateStart.UTC().Format(layout), debateEnd.UTC().Format(layout),
|
|
||||||
schema,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
func trimRightSlash(s string) string {
|
|
||||||
for len(s) > 0 && s[len(s)-1] == '/' {
|
|
||||||
s = s[:len(s)-1]
|
|
||||||
}
|
|
||||||
return s
|
|
||||||
}
|
|
||||||
@@ -109,12 +109,6 @@ type createTopicBody struct {
|
|||||||
SignupCloseAt string `json:"signup_close_at"`
|
SignupCloseAt string `json:"signup_close_at"`
|
||||||
DebateStartAt string `json:"debate_start_at"`
|
DebateStartAt string `json:"debate_start_at"`
|
||||||
DebateEndAt string `json:"debate_end_at"`
|
DebateEndAt string `json:"debate_end_at"`
|
||||||
// Optional: per-topic announce-channel target. Both must be set
|
|
||||||
// (or both omitted = no broadcasts). Creator picks based on the
|
|
||||||
// debate's intended audience (different guilds may host different
|
|
||||||
// communities, different channels may serve different categories).
|
|
||||||
AnnounceGuildBaseURL string `json:"announce_guild_base_url,omitempty"`
|
|
||||||
AnnounceChannelID string `json:"announce_channel_id,omitempty"`
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// POST /api/topics
|
// POST /api/topics
|
||||||
@@ -147,19 +141,6 @@ func (h *TopicsHandler) Create(w http.ResponseWriter, r *http.Request) {
|
|||||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// Announce target: both fields required or both empty; one-of-two
|
|
||||||
// is a config error caught here rather than silently treated as
|
|
||||||
// "no broadcast".
|
|
||||||
var aGuild, aChannel *string
|
|
||||||
if body.AnnounceGuildBaseURL != "" || body.AnnounceChannelID != "" {
|
|
||||||
if body.AnnounceGuildBaseURL == "" || body.AnnounceChannelID == "" {
|
|
||||||
http.Error(w, "announce_guild_base_url and announce_channel_id must both be set (or both empty for no broadcasts)", http.StatusBadRequest)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
g, c := body.AnnounceGuildBaseURL, body.AnnounceChannelID
|
|
||||||
aGuild, aChannel = &g, &c
|
|
||||||
}
|
|
||||||
|
|
||||||
created, err := h.store.Create(r.Context(), store.CreateTopicInput{
|
created, err := h.store.Create(r.Context(), store.CreateTopicInput{
|
||||||
Title: body.Title,
|
Title: body.Title,
|
||||||
Summary: body.Summary,
|
Summary: body.Summary,
|
||||||
@@ -170,8 +151,6 @@ func (h *TopicsHandler) Create(w http.ResponseWriter, r *http.Request) {
|
|||||||
DebateStartAt: parsed[2],
|
DebateStartAt: parsed[2],
|
||||||
DebateEndAt: parsed[3],
|
DebateEndAt: parsed[3],
|
||||||
CreatorUserID: caller.ID,
|
CreatorUserID: caller.ID,
|
||||||
AnnounceGuildBaseURL: aGuild,
|
|
||||||
AnnounceChannelID: aChannel,
|
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
http.Error(w, "create failed: "+err.Error(), http.StatusInternalServerError)
|
http.Error(w, "create failed: "+err.Error(), http.StatusInternalServerError)
|
||||||
|
|||||||
@@ -1,17 +1,14 @@
|
|||||||
package handlers
|
package handlers
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
|
||||||
"net/http"
|
"net/http"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/go-chi/chi/v5"
|
"github.com/go-chi/chi/v5"
|
||||||
|
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/auth"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/auth"
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/fabric"
|
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/models"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/models"
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/store"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/store"
|
||||||
)
|
)
|
||||||
@@ -20,11 +17,10 @@ type VerdictHandler struct {
|
|||||||
topics *store.TopicStore
|
topics *store.TopicStore
|
||||||
camps *store.CampStore
|
camps *store.CampStore
|
||||||
verdicts *store.VerdictStore
|
verdicts *store.VerdictStore
|
||||||
announcer *fabric.Announcer // optional; nil-safe via Enabled() check
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewVerdictHandler(t *store.TopicStore, c *store.CampStore, v *store.VerdictStore, ann *fabric.Announcer) *VerdictHandler {
|
func NewVerdictHandler(t *store.TopicStore, c *store.CampStore, v *store.VerdictStore) *VerdictHandler {
|
||||||
return &VerdictHandler{topics: t, camps: c, verdicts: v, announcer: ann}
|
return &VerdictHandler{topics: t, camps: c, verdicts: v}
|
||||||
}
|
}
|
||||||
|
|
||||||
type submitVerdictBody struct {
|
type submitVerdictBody struct {
|
||||||
@@ -117,24 +113,6 @@ func (h *VerdictHandler) Submit(w http.ResponseWriter, r *http.Request) {
|
|||||||
w.Header().Set("x-warn", "verdict saved but status update failed: "+err.Error())
|
w.Header().Set("x-warn", "verdict saved but status update failed: "+err.Error())
|
||||||
}
|
}
|
||||||
|
|
||||||
// Lifecycle broadcast — completed event. Best-effort; runs async
|
|
||||||
// outside the request context so a slow Fabric doesn't slow the
|
|
||||||
// judge's response. Target is per-topic (nil announcer or nil
|
|
||||||
// target on topic → silently skipped by announcer).
|
|
||||||
if h.announcer != nil {
|
|
||||||
var tgt fabric.Target
|
|
||||||
if topic.AnnounceGuildBaseURL != nil && topic.AnnounceChannelID != nil {
|
|
||||||
tgt = fabric.Target{
|
|
||||||
GuildBaseURL: *topic.AnnounceGuildBaseURL,
|
|
||||||
ChannelID: *topic.AnnounceChannelID,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
go func(t *fabric.Announcer, tID, title, judge string, target fabric.Target) {
|
|
||||||
summary := fmt.Sprintf("verdict published by judge=%s. Use dialectic_view_verdict to see the structured result.", judge)
|
|
||||||
_ = t.PostLifecycleEvent(context.Background(), target, tID, title, "completed", summary)
|
|
||||||
}(h.announcer, verdict.TopicID, topic.Title, caller.ID, tgt)
|
|
||||||
}
|
|
||||||
|
|
||||||
writeJSON(w, http.StatusCreated, map[string]any{
|
writeJSON(w, http.StatusCreated, map[string]any{
|
||||||
"id": verdict.ID,
|
"id": verdict.ID,
|
||||||
"topic_id": verdict.TopicID,
|
"topic_id": verdict.TopicID,
|
||||||
|
|||||||
@@ -11,7 +11,6 @@ import (
|
|||||||
|
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/auth"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/auth"
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/config"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/config"
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/fabric"
|
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/httpapi/handlers"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/httpapi/handlers"
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/store"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/store"
|
||||||
)
|
)
|
||||||
@@ -63,8 +62,7 @@ func Mount(cfg *config.Config, db *sqlx.DB, version string) http.Handler {
|
|||||||
topicsH := handlers.NewTopicsHandler(topicStore, campStore)
|
topicsH := handlers.NewTopicsHandler(topicStore, campStore)
|
||||||
signupsH := handlers.NewSignupsHandler(topicStore, signupStore)
|
signupsH := handlers.NewSignupsHandler(topicStore, signupStore)
|
||||||
argsH := handlers.NewArgumentsHandler(topicStore, campStore, roundStore, argStore)
|
argsH := handlers.NewArgumentsHandler(topicStore, campStore, roundStore, argStore)
|
||||||
announcer := fabric.NewAnnouncer(cfg.FabricSystemAPIKey)
|
verdictH := handlers.NewVerdictHandler(topicStore, campStore, verdictStore)
|
||||||
verdictH := handlers.NewVerdictHandler(topicStore, campStore, verdictStore, announcer)
|
|
||||||
adminH := handlers.NewAdminHandler(db, cfg.AgentAPIKeyPepper, cfg.DialecticAdminAPIKey)
|
adminH := handlers.NewAdminHandler(db, cfg.AgentAPIKeyPepper, cfg.DialecticAdminAPIKey)
|
||||||
|
|
||||||
// Routes.
|
// Routes.
|
||||||
|
|||||||
@@ -49,13 +49,6 @@ type Topic struct {
|
|||||||
VisibilityChangedBy *string `db:"visibility_changed_by" json:"visibility_changed_by,omitempty"`
|
VisibilityChangedBy *string `db:"visibility_changed_by" json:"visibility_changed_by,omitempty"`
|
||||||
VisibilityChangedAt *time.Time `db:"visibility_changed_at" json:"visibility_changed_at,omitempty"`
|
VisibilityChangedAt *time.Time `db:"visibility_changed_at" json:"visibility_changed_at,omitempty"`
|
||||||
CancelledReason *string `db:"cancelled_reason" json:"cancelled_reason,omitempty"`
|
CancelledReason *string `db:"cancelled_reason" json:"cancelled_reason,omitempty"`
|
||||||
// AnnounceGuildBaseURL + AnnounceChannelID: per-topic broadcast
|
|
||||||
// target. Set by the proposing agent at create time (they pick which
|
|
||||||
// Fabric guild + announce channel reflects this topic's audience).
|
|
||||||
// Both nullable; null on either side disables lifecycle broadcasts
|
|
||||||
// for this topic (creator opted out).
|
|
||||||
AnnounceGuildBaseURL *string `db:"announce_guild_base_url" json:"announce_guild_base_url,omitempty"`
|
|
||||||
AnnounceChannelID *string `db:"announce_channel_id" json:"announce_channel_id,omitempty"`
|
|
||||||
CreatedAt time.Time `db:"created_at" json:"created_at"`
|
CreatedAt time.Time `db:"created_at" json:"created_at"`
|
||||||
UpdatedAt time.Time `db:"updated_at" json:"updated_at"`
|
UpdatedAt time.Time `db:"updated_at" json:"updated_at"`
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,14 +2,12 @@ package orchestrator
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
|
||||||
"log"
|
"log"
|
||||||
"math/rand"
|
"math/rand"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/jmoiron/sqlx"
|
"github.com/jmoiron/sqlx"
|
||||||
|
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/fabric"
|
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/models"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/models"
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/store"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/store"
|
||||||
)
|
)
|
||||||
@@ -21,7 +19,6 @@ import (
|
|||||||
// State transitions handled by the ticker:
|
// State transitions handled by the ticker:
|
||||||
//
|
//
|
||||||
// created → signup_open (when now >= signup_open_at)
|
// created → signup_open (when now >= signup_open_at)
|
||||||
// + post Fabric announcement
|
|
||||||
// signup_open → signup_closed (when now >= signup_close_at, allocator succeeded)
|
// signup_open → signup_closed (when now >= signup_close_at, allocator succeeded)
|
||||||
// → cancelled (allocator returned CancelReason)
|
// → cancelled (allocator returned CancelReason)
|
||||||
// signup_closed → debating (when now >= debate_start_at; opens round 0)
|
// signup_closed → debating (when now >= debate_start_at; opens round 0)
|
||||||
@@ -35,13 +32,19 @@ import (
|
|||||||
//
|
//
|
||||||
// Per-topic transitions use SELECT FOR UPDATE so concurrent ticker
|
// Per-topic transitions use SELECT FOR UPDATE so concurrent ticker
|
||||||
// instances (or future replicas) don't double-fire.
|
// instances (or future replicas) don't double-fire.
|
||||||
|
//
|
||||||
|
// Lifecycle broadcasting moved out-of-backend (Aug 2026): the proposing
|
||||||
|
// agent posts a single recruitment fabric-send-message after creating a
|
||||||
|
// topic; downstream agents book HF on_call slots covering the debate
|
||||||
|
// window via `hf calendar schedule`, and HF wakes them naturally. The
|
||||||
|
// backend stays a pure data + state-machine service and doesn't know
|
||||||
|
// about Fabric.
|
||||||
type Ticker struct {
|
type Ticker struct {
|
||||||
db *sqlx.DB
|
db *sqlx.DB
|
||||||
topics *store.TopicStore
|
topics *store.TopicStore
|
||||||
signups *store.SignupStore
|
signups *store.SignupStore
|
||||||
camps *store.CampStore
|
camps *store.CampStore
|
||||||
rounds *store.RoundStore
|
rounds *store.RoundStore
|
||||||
announcer *fabric.Announcer
|
|
||||||
interval time.Duration
|
interval time.Duration
|
||||||
rng *rand.Rand
|
rng *rand.Rand
|
||||||
}
|
}
|
||||||
@@ -52,7 +55,6 @@ func NewTicker(
|
|||||||
signups *store.SignupStore,
|
signups *store.SignupStore,
|
||||||
camps *store.CampStore,
|
camps *store.CampStore,
|
||||||
rounds *store.RoundStore,
|
rounds *store.RoundStore,
|
||||||
announcer *fabric.Announcer,
|
|
||||||
interval time.Duration,
|
interval time.Duration,
|
||||||
) *Ticker {
|
) *Ticker {
|
||||||
if interval <= 0 {
|
if interval <= 0 {
|
||||||
@@ -64,7 +66,6 @@ func NewTicker(
|
|||||||
signups: signups,
|
signups: signups,
|
||||||
camps: camps,
|
camps: camps,
|
||||||
rounds: rounds,
|
rounds: rounds,
|
||||||
announcer: announcer,
|
|
||||||
interval: interval,
|
interval: interval,
|
||||||
rng: rand.New(rand.NewSource(time.Now().UnixNano())),
|
rng: rand.New(rand.NewSource(time.Now().UnixNano())),
|
||||||
}
|
}
|
||||||
@@ -72,7 +73,7 @@ func NewTicker(
|
|||||||
|
|
||||||
// Run blocks until ctx is cancelled. Caller goroutines it.
|
// Run blocks until ctx is cancelled. Caller goroutines it.
|
||||||
func (t *Ticker) Run(ctx context.Context) {
|
func (t *Ticker) Run(ctx context.Context) {
|
||||||
log.Printf("orchestrator: ticker started (interval=%s, announce=%v)", t.interval, t.announcer.Enabled())
|
log.Printf("orchestrator: ticker started (interval=%s)", t.interval)
|
||||||
tk := time.NewTicker(t.interval)
|
tk := time.NewTicker(t.interval)
|
||||||
defer tk.Stop()
|
defer tk.Stop()
|
||||||
// First tick immediately so startup is responsive — don't wait
|
// First tick immediately so startup is responsive — don't wait
|
||||||
@@ -81,7 +82,7 @@ func (t *Ticker) Run(ctx context.Context) {
|
|||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
log.Printf("orchestrator: ticker stopped")
|
log.Printf("orchestrator: ticker stopping")
|
||||||
return
|
return
|
||||||
case <-tk.C:
|
case <-tk.C:
|
||||||
t.tickOnce(ctx)
|
t.tickOnce(ctx)
|
||||||
@@ -89,8 +90,6 @@ func (t *Ticker) Run(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// tickOnce scans + applies. Errors are logged per topic; one topic
|
|
||||||
// failing doesn't stall the others.
|
|
||||||
func (t *Ticker) tickOnce(ctx context.Context) {
|
func (t *Ticker) tickOnce(ctx context.Context) {
|
||||||
now := time.Now()
|
now := time.Now()
|
||||||
|
|
||||||
@@ -98,17 +97,11 @@ func (t *Ticker) tickOnce(ctx context.Context) {
|
|||||||
if err := t.transitionByStatus(ctx, now,
|
if err := t.transitionByStatus(ctx, now,
|
||||||
models.TopicStatusCreated, "signup_open_at",
|
models.TopicStatusCreated, "signup_open_at",
|
||||||
func(ctx context.Context, tx *sqlx.Tx, topicID string) error {
|
func(ctx context.Context, tx *sqlx.Tx, topicID string) error {
|
||||||
topic, err := t.topics.GetByID(ctx, topicID)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if _, err := tx.ExecContext(ctx,
|
if _, err := tx.ExecContext(ctx,
|
||||||
`UPDATE topics SET status = ? WHERE id = ?`,
|
`UPDATE topics SET status = ? WHERE id = ?`,
|
||||||
models.TopicStatusSignupOpen, topicID); err != nil {
|
models.TopicStatusSignupOpen, topicID); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
// Announcement is best-effort, outside the tx (network call).
|
|
||||||
go t.broadcastAnnouncement(topic)
|
|
||||||
return nil
|
return nil
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
log.Printf("orchestrator: created→signup_open scan: %v", err)
|
log.Printf("orchestrator: created→signup_open scan: %v", err)
|
||||||
@@ -118,10 +111,6 @@ func (t *Ticker) tickOnce(ctx context.Context) {
|
|||||||
if err := t.transitionByStatus(ctx, now,
|
if err := t.transitionByStatus(ctx, now,
|
||||||
models.TopicStatusSignupOpen, "signup_close_at",
|
models.TopicStatusSignupOpen, "signup_close_at",
|
||||||
func(ctx context.Context, tx *sqlx.Tx, topicID string) error {
|
func(ctx context.Context, tx *sqlx.Tx, topicID string) error {
|
||||||
topic, err := t.topics.GetByID(ctx, topicID)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
signups, err := t.signups.ListByTopic(ctx, topicID)
|
signups, err := t.signups.ListByTopic(ctx, topicID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -133,10 +122,6 @@ func (t *Ticker) tickOnce(ctx context.Context) {
|
|||||||
models.TopicStatusCancelled, res.CancelReason, topicID)
|
models.TopicStatusCancelled, res.CancelReason, topicID)
|
||||||
log.Printf("orchestrator: topic %s cancelled at signup_close: %s",
|
log.Printf("orchestrator: topic %s cancelled at signup_close: %s",
|
||||||
topicID, res.CancelReason)
|
topicID, res.CancelReason)
|
||||||
if err == nil {
|
|
||||||
go t.broadcastLifecycle(topic, "cancelled",
|
|
||||||
fmt.Sprintf("debate cancelled at signup close - %s", res.CancelReason))
|
|
||||||
}
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if err := t.camps.WriteAllocation(ctx, tx, topicID, res.Allocation); err != nil {
|
if err := t.camps.WriteAllocation(ctx, tx, topicID, res.Allocation); err != nil {
|
||||||
@@ -148,14 +133,6 @@ func (t *Ticker) tickOnce(ctx context.Context) {
|
|||||||
log.Printf("orchestrator: topic %s allocated pro=%s con=%s judge=%s",
|
log.Printf("orchestrator: topic %s allocated pro=%s con=%s judge=%s",
|
||||||
topicID,
|
topicID,
|
||||||
res.Allocation[models.CampPro], res.Allocation[models.CampCon], res.Allocation[models.CampJudge])
|
res.Allocation[models.CampPro], res.Allocation[models.CampCon], res.Allocation[models.CampJudge])
|
||||||
if err == nil {
|
|
||||||
go t.broadcastLifecycle(topic, "signup_closed",
|
|
||||||
fmt.Sprintf("camps allocated — pro=%s con=%s judge=%s. Debate starts at %s",
|
|
||||||
res.Allocation[models.CampPro],
|
|
||||||
res.Allocation[models.CampCon],
|
|
||||||
res.Allocation[models.CampJudge],
|
|
||||||
topic.DebateStartAt.UTC().Format("2006-01-02 15:04 UTC")))
|
|
||||||
}
|
|
||||||
return err
|
return err
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
log.Printf("orchestrator: signup_open→signup_closed scan: %v", err)
|
log.Printf("orchestrator: signup_open→signup_closed scan: %v", err)
|
||||||
@@ -165,10 +142,6 @@ func (t *Ticker) tickOnce(ctx context.Context) {
|
|||||||
if err := t.transitionByStatus(ctx, now,
|
if err := t.transitionByStatus(ctx, now,
|
||||||
models.TopicStatusSignupClosed, "debate_start_at",
|
models.TopicStatusSignupClosed, "debate_start_at",
|
||||||
func(ctx context.Context, tx *sqlx.Tx, topicID string) error {
|
func(ctx context.Context, tx *sqlx.Tx, topicID string) error {
|
||||||
topic, err := t.topics.GetByID(ctx, topicID)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
if _, err := tx.ExecContext(ctx,
|
if _, err := tx.ExecContext(ctx,
|
||||||
`UPDATE topics SET status = ? WHERE id = ?`,
|
`UPDATE topics SET status = ? WHERE id = ?`,
|
||||||
models.TopicStatusDebating, topicID); err != nil {
|
models.TopicStatusDebating, topicID); err != nil {
|
||||||
@@ -176,15 +149,10 @@ func (t *Ticker) tickOnce(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
// Round 0 inserted within the tx — if commit fails we don't
|
// Round 0 inserted within the tx — if commit fails we don't
|
||||||
// leak a half-state.
|
// leak a half-state.
|
||||||
_, err = tx.ExecContext(ctx,
|
_, err := tx.ExecContext(ctx,
|
||||||
`INSERT INTO rounds (id, topic_id, round_no) VALUES (UUID(), ?, 0)`,
|
`INSERT INTO rounds (id, topic_id, round_no) VALUES (UUID(), ?, 0)`,
|
||||||
topicID)
|
topicID)
|
||||||
log.Printf("orchestrator: topic %s entered debating; round 0 opened", topicID)
|
log.Printf("orchestrator: topic %s entered debating; round 0 opened", topicID)
|
||||||
if err == nil {
|
|
||||||
go t.broadcastLifecycle(topic, "debating",
|
|
||||||
fmt.Sprintf("debate is live — pro/con post arguments; judge stays mostly silent until debate_end_at (%s). Use participate-debate workflow.",
|
|
||||||
topic.DebateEndAt.UTC().Format("2006-01-02 15:04 UTC")))
|
|
||||||
}
|
|
||||||
return err
|
return err
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
log.Printf("orchestrator: signup_closed→debating scan: %v", err)
|
log.Printf("orchestrator: signup_closed→debating scan: %v", err)
|
||||||
@@ -193,10 +161,7 @@ func (t *Ticker) tickOnce(ctx context.Context) {
|
|||||||
// Note: there's no explicit `debating → judging` transition in v1.
|
// Note: there's no explicit `debating → judging` transition in v1.
|
||||||
// The verdict handler enforces "status==debating AND now>=debate_end_at"
|
// The verdict handler enforces "status==debating AND now>=debate_end_at"
|
||||||
// as its preconditions; that's equivalent to a "judging" gate without
|
// as its preconditions; that's equivalent to a "judging" gate without
|
||||||
// adding a new enum value. Migration 002 will introduce the explicit
|
// adding a new enum value.
|
||||||
// 'judging' state when we want richer UI (e.g. "Awaiting verdict"
|
|
||||||
// distinct from "In debate"); until then this comment serves as the
|
|
||||||
// state-machine documentation for future maintainers.
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// transitionByStatus is the shared "scan + per-row tx + apply" pattern.
|
// transitionByStatus is the shared "scan + per-row tx + apply" pattern.
|
||||||
@@ -245,48 +210,3 @@ func (t *Ticker) applyOne(ctx context.Context, topicID string,
|
|||||||
}
|
}
|
||||||
return tx.Commit()
|
return tx.Commit()
|
||||||
}
|
}
|
||||||
|
|
||||||
// broadcastLifecycle wraps the announcer's lifecycle-event post with
|
|
||||||
// the standard signup_closed / cancelled / debating / completed
|
|
||||||
// formats. Best-effort; runs in its own goroutine outside any tx.
|
|
||||||
// Target is resolved from the topic's per-topic announce columns;
|
|
||||||
// null on either column → announcer skips with a log (creator opted
|
|
||||||
// out of broadcasts).
|
|
||||||
func (t *Ticker) broadcastLifecycle(topic *models.Topic, kind, summary string) {
|
|
||||||
if topic == nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if err := t.announcer.PostLifecycleEvent(
|
|
||||||
context.Background(), topicTarget(topic), topic.ID, topic.Title, kind, summary,
|
|
||||||
); err != nil {
|
|
||||||
log.Printf("orchestrator: lifecycle broadcast topic=%s kind=%s failed: %v", topic.ID, kind, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (t *Ticker) broadcastAnnouncement(topic *models.Topic) {
|
|
||||||
if topic == nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if err := t.announcer.PostTopicAnnouncement(
|
|
||||||
context.Background(), topicTarget(topic),
|
|
||||||
topic.ID, topic.Title, topic.Summary,
|
|
||||||
topic.SignupOpenAt, topic.SignupCloseAt,
|
|
||||||
topic.DebateStartAt, topic.DebateEndAt,
|
|
||||||
topic.VerdictSchemaID,
|
|
||||||
); err != nil {
|
|
||||||
log.Printf("orchestrator: announce topic=%s failed: %v", topic.ID, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// topicTarget extracts the per-topic announce target from the topic
|
|
||||||
// row; returns zero-value Target if either column is null (which the
|
|
||||||
// announcer treats as "skip").
|
|
||||||
func topicTarget(topic *models.Topic) fabric.Target {
|
|
||||||
if topic.AnnounceGuildBaseURL == nil || topic.AnnounceChannelID == nil {
|
|
||||||
return fabric.Target{}
|
|
||||||
}
|
|
||||||
return fabric.Target{
|
|
||||||
GuildBaseURL: *topic.AnnounceGuildBaseURL,
|
|
||||||
ChannelID: *topic.AnnounceChannelID,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -32,8 +32,6 @@ type CreateTopicInput struct {
|
|||||||
DebateStartAt time.Time
|
DebateStartAt time.Time
|
||||||
DebateEndAt time.Time
|
DebateEndAt time.Time
|
||||||
CreatorUserID string
|
CreatorUserID string
|
||||||
AnnounceGuildBaseURL *string // optional; null = no broadcasts for this topic
|
|
||||||
AnnounceChannelID *string
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *TopicStore) Create(ctx context.Context, in CreateTopicInput) (*models.Topic, error) {
|
func (s *TopicStore) Create(ctx context.Context, in CreateTopicInput) (*models.Topic, error) {
|
||||||
@@ -41,11 +39,11 @@ func (s *TopicStore) Create(ctx context.Context, in CreateTopicInput) (*models.T
|
|||||||
_, err := s.db.ExecContext(ctx, `
|
_, err := s.db.ExecContext(ctx, `
|
||||||
INSERT INTO topics (id, title, summary, visibility, verdict_schema_id,
|
INSERT INTO topics (id, title, summary, visibility, verdict_schema_id,
|
||||||
signup_open_at, signup_close_at, debate_start_at, debate_end_at,
|
signup_open_at, signup_close_at, debate_start_at, debate_end_at,
|
||||||
creator_user_id, announce_guild_base_url, announce_channel_id)
|
creator_user_id)
|
||||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||||
id, in.Title, in.Summary, in.Visibility, in.VerdictSchemaID,
|
id, in.Title, in.Summary, in.Visibility, in.VerdictSchemaID,
|
||||||
in.SignupOpenAt, in.SignupCloseAt, in.DebateStartAt, in.DebateEndAt,
|
in.SignupOpenAt, in.SignupCloseAt, in.DebateStartAt, in.DebateEndAt,
|
||||||
in.CreatorUserID, in.AnnounceGuildBaseURL, in.AnnounceChannelID)
|
in.CreatorUserID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("insert topic: %w", err)
|
return nil, fmt.Errorf("insert topic: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
9
main.go
9
main.go
@@ -19,7 +19,6 @@ import (
|
|||||||
|
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/config"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/config"
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/db"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/db"
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/fabric"
|
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/httpapi"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/httpapi"
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/orchestrator"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/orchestrator"
|
||||||
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/store"
|
"git.hangman-lab.top/hzhang/Dialectic.Backend/internal/store"
|
||||||
@@ -51,15 +50,15 @@ func main() {
|
|||||||
}
|
}
|
||||||
log.Printf("migrations: ok")
|
log.Printf("migrations: ok")
|
||||||
|
|
||||||
// Wire orchestrator + Fabric announcer + start the ticker.
|
// Wire orchestrator + start the ticker. Backend no longer broadcasts
|
||||||
// Per Phase 3.5: target is per-topic, only the system key stays in env.
|
// to Fabric — proposers post a single recruitment fabric-send-message,
|
||||||
announcer := fabric.NewAnnouncer(cfg.FabricSystemAPIKey)
|
// downstream agents book HF on_call slots to be woken at debate time.
|
||||||
topicStore := store.NewTopicStore(conn)
|
topicStore := store.NewTopicStore(conn)
|
||||||
signupStore := store.NewSignupStore(conn)
|
signupStore := store.NewSignupStore(conn)
|
||||||
campStore := store.NewCampStore(conn)
|
campStore := store.NewCampStore(conn)
|
||||||
roundStore := store.NewRoundStore(conn)
|
roundStore := store.NewRoundStore(conn)
|
||||||
ticker := orchestrator.NewTicker(conn, topicStore, signupStore, campStore, roundStore,
|
ticker := orchestrator.NewTicker(conn, topicStore, signupStore, campStore, roundStore,
|
||||||
announcer, cfg.OrchestratorTickInterval)
|
cfg.OrchestratorTickInterval)
|
||||||
go ticker.Run(ctx)
|
go ticker.Run(ctx)
|
||||||
|
|
||||||
srv := &http.Server{
|
srv := &http.Server{
|
||||||
|
|||||||
Reference in New Issue
Block a user