From 4093bdb68384d0ca6f643ac989b47a3c91ce5195 Mon Sep 17 00:00:00 2001
From: kami <kami@kvmx.ru>
Date: Sun, 26 Jul 2026 20:31:42 +0400
Subject: [PATCH] add telegram and ntfy event delivery

---
 cmd/orchestra/main.go         |  16 +++++
 internal/delivery/delivery.go | 129 ++++++++++++++++++++++++++++++++++
 progress.md                   |   2 +
 3 files changed, 147 insertions(+)
 create mode 100644 internal/delivery/delivery.go

diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go
index 2eb61f7..a3a7736 100644
--- a/cmd/orchestra/main.go
+++ b/cmd/orchestra/main.go
@@ -8,6 +8,7 @@ import (
 	"log"
 	"net/http"
 	"orchestra/internal/authz"
+	"orchestra/internal/delivery"
 	"orchestra/internal/domain"
 	"orchestra/internal/herdr"
 	"orchestra/internal/operations"
@@ -356,6 +357,21 @@ func main() {
 		sup.Start(context.Background())
 		providerHealth["jsonl"] = sup
 	}
+	var senders []delivery.Sender
+	if token, chat := os.Getenv("ORCHESTRA_TELEGRAM_BOT_TOKEN"), os.Getenv("ORCHESTRA_TELEGRAM_CHAT_ID"); token != "" && chat != "" {
+		senders = append(senders, delivery.Telegram{Token: token, ChatID: chat})
+	}
+	if topic := os.Getenv("ORCHESTRA_NTFY_TOPIC"); topic != "" {
+		senders = append(senders, delivery.Ntfy{Topic: topic, Token: os.Getenv("ORCHESTRA_NTFY_TOKEN"), URL: os.Getenv("ORCHESTRA_NTFY_URL")})
+	}
+	if len(senders) > 0 {
+		go func() {
+			err := (&delivery.Fanout{Senders: senders}).Run(context.Background(), s.Events)
+			if err != nil {
+				log.Printf("delivery fanout: %v", err)
+			}
+		}()
+	}
 	port := os.Getenv("ORCHESTRA_PORT")
 	if port == "" {
 		port = "9145"
diff --git a/internal/delivery/delivery.go b/internal/delivery/delivery.go
new file mode 100644
index 0000000..05afce5
--- /dev/null
+++ b/internal/delivery/delivery.go
@@ -0,0 +1,129 @@
+package delivery
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"orchestra/internal/domain"
+	"time"
+)
+
+type Sender interface {
+	Send(context.Context, string) error
+}
+
+type Telegram struct {
+	Token, ChatID string
+	Client        *http.Client
+}
+
+func (t Telegram) Send(ctx context.Context, message string) error {
+	if t.Token == "" || t.ChatID == "" {
+		return fmt.Errorf("telegram credentials missing")
+	}
+	b, _ := json.Marshal(map[string]string{"chat_id": t.ChatID, "text": message})
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://api.telegram.org/bot"+t.Token+"/sendMessage", bytes.NewReader(b))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	c := t.Client
+	if c == nil {
+		c = http.DefaultClient
+	}
+	resp, err := c.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode/100 != 2 {
+		return fmt.Errorf("telegram: %s", resp.Status)
+	}
+	return nil
+}
+
+type Ntfy struct {
+	Topic, Token string
+	Client       *http.Client
+	URL          string
+}
+
+func (n Ntfy) Send(ctx context.Context, message string) error {
+	if n.Topic == "" {
+		return fmt.Errorf("ntfy topic missing")
+	}
+	u := n.URL
+	if u == "" {
+		u = "https://ntfy.sh/" + n.Topic
+	} else {
+		u += "/" + n.Topic
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewBufferString(message))
+	if err != nil {
+		return err
+	}
+	if n.Token != "" {
+		req.Header.Set("Authorization", "Bearer "+n.Token)
+	}
+	c := n.Client
+	if c == nil {
+		c = http.DefaultClient
+	}
+	resp, err := c.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode/100 != 2 {
+		return fmt.Errorf("ntfy: %s", resp.Status)
+	}
+	return nil
+}
+
+func Message(e domain.Event) (string, bool) {
+	switch e.Type {
+	case "TaskCompleted":
+		return "Task completed: " + e.TaskID, true
+	case "TaskFailed":
+		return "Task failed: " + e.TaskID, true
+	case "TaskBlocked":
+		return "Task blocked: " + e.TaskID, true
+	case "ApprovalRequested":
+		return "Approval requested: " + e.TaskID, true
+	default:
+		return "", false
+	}
+}
+
+type Fanout struct {
+	Senders []Sender
+	Cursor  uint64
+	Retry   time.Duration
+}
+
+func (f *Fanout) Run(ctx context.Context, events func(uint64) []domain.Event) error {
+	if f.Retry <= 0 {
+		f.Retry = 5 * time.Second
+	}
+	for {
+		for _, e := range events(f.Cursor) {
+			if msg, ok := Message(e); ok {
+				for _, s := range f.Senders {
+					if err := s.Send(ctx, msg); err != nil {
+						return err
+					}
+				}
+			}
+			if e.Seq > f.Cursor {
+				f.Cursor = e.Seq
+			}
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(f.Retry):
+		}
+	}
+}
diff --git a/progress.md b/progress.md
index d6f8854..513a9be 100644
--- a/progress.md
+++ b/progress.md
@@ -65,6 +65,8 @@ Provider supervision/reflection is now wired: Gitea webhook and polling use appe
 
 Quota reporting now has strict payload validation, and `router.QuotaAvailability` implements rolling-window conservative headroom filtering: a harness is considered full at 80% of its configured limit. Standup event payloads also require an items field; scheduled advisory production and approval application remain open.
 
+Notification delivery now supports Telegram and ntfy fan-out for completion, failure, block, and approval events, with event cursors, bounded polling, and notify-only surface policy preserved. Configure `ORCHESTRA_TELEGRAM_BOT_TOKEN`/`ORCHESTRA_TELEGRAM_CHAT_ID` or `ORCHESTRA_NTFY_TOPIC` to enable it.
+
 Recommended order:
 
 1. Add the orchestration coordinator: lease → worktree → harness session → bootstrap → lifecycle events.