feat: tier pushes by urgency and group them per session

apps/relay/src/index.test.ts

@@ -58,6 +58,14 @@ describe('POST /push', () => {
     expect(payload.aps['mutable-content']).toBe(1);
   });
 
+  test('tiers the APNs envelope from level and threadKey', async () => {
+    const { sent, fetchImpl } = apnsStub(ok);
+    await push(fakeEnv(pem), fetchImpl, { token: TOKEN, ciphertext: 'c2VjcmV0', level: 'urgent', threadKey: 'abc123' });
+    const { aps } = JSON.parse(String(sent[0]?.init.body));
+    expect(aps['interruption-level']).toBe('time-sensitive');
+    expect(aps['thread-id']).toBe('abc123');
+  });
+
   test('retries a BadDeviceToken once on the other environment', async () => {
     const { sent, fetchImpl } = apnsStub((url) =>
       url.startsWith('https://api.push.apple.com')

apps/relay/src/payload.test.ts

@@ -25,6 +25,21 @@ describe('pushRequestSchema', () => {
     expect(pushRequestSchema.safeParse({ token: TOKEN }).success).toBe(false);
   });
 
+  test('an absent or unknown level parses as normal so older callers keep working', () => {
+    for (const level of [undefined, 'loud', '', 7]) {
+      const r = pushRequestSchema.safeParse({ token: TOKEN, ciphertext: 'x', level });
+      expect(r.success && r.data.level).toBe('normal');
+    }
+    expect(pushRequestSchema.safeParse({ token: TOKEN, ciphertext: 'x', level: 'urgent' }).data?.level).toBe('urgent');
+    expect(pushRequestSchema.safeParse({ token: TOKEN, ciphertext: 'x', level: 'quiet' }).data?.level).toBe('quiet');
+  });
+
+  test('a malformed thread key is dropped rather than rejecting the push', () => {
+    const r = pushRequestSchema.safeParse({ token: TOKEN, ciphertext: 'x', threadKey: 'my session!' });
+    expect(r.success && r.data.threadKey).toBeUndefined();
+    expect(pushRequestSchema.safeParse({ token: TOKEN, ciphertext: 'x', threadKey: 'ab12' }).data?.threadKey).toBe('ab12');
+  });
+
   test.each([
     ['too short', 'a'.repeat(63)],
     ['too long', 'a'.repeat(65)],
@@ -51,6 +66,32 @@ describe('buildApnsPayload', () => {
     expect(payload.e).toBeUndefined();
   });
 
+  test('urgent pushes are time-sensitive with full relevance and a sound', () => {
+    const { aps } = buildApnsPayload({ token: TOKEN, ciphertext: 'x', level: 'urgent' });
+    expect(aps['interruption-level']).toBe('time-sensitive');
+    expect(aps['relevance-score']).toBe(1);
+    expect(aps.sound).toBe('default');
+  });
+
+  test('normal pushes are active with a sound, and the default when level is absent', () => {
+    for (const level of ['normal', undefined] as const) {
+      const { aps } = buildApnsPayload({ token: TOKEN, ciphertext: 'x', level });
+      expect(aps['interruption-level']).toBe('active');
+      expect(aps.sound).toBe('default');
+    }
+  });
+
+  test('quiet pushes are passive and silent', () => {
+    const { aps } = buildApnsPayload({ token: TOKEN, body: 'x', level: 'quiet' });
+    expect(aps['interruption-level']).toBe('passive');
+    expect(aps.sound).toBeUndefined();
+  });
+
+  test('thread-id is set from the opaque key only when given', () => {
+    expect(buildApnsPayload({ token: TOKEN, ciphertext: 'x', threadKey: 'abc123' }).aps['thread-id']).toBe('abc123');
+    expect(buildApnsPayload({ token: TOKEN, ciphertext: 'x' }).aps['thread-id']).toBeUndefined();
+  });
+
   test('deep link rides alongside the payload when present', () => {
     const payload = buildApnsPayload({ token: TOKEN, body: 'x', deepLink: 'tether://session/a' });
     expect(payload.link).toBe('tether://session/a');

apps/relay/src/payload.ts

@@ -2,6 +2,8 @@ import { z } from 'zod';
 
 // A push is either Phase 1 (cleartext) or Phase 2 (ciphertext the NSE decrypts);
 // never both, or a caller leaked readable text alongside the encrypted copy.
+export type PushLevel = 'urgent' | 'normal' | 'quiet';
+
 export const pushRequestSchema = z
   .object({
     token: z.string().regex(/^[0-9a-fA-F]{64}$/, 'token must be a 64-char hex APNs device token'),
@@ -10,6 +12,17 @@ export const pushRequestSchema = z
     ciphertext: z.string().min(1).max(3000).optional(),
     collapseId: z.string().min(1).max(64).optional(),
     deepLink: z.string().max(500).optional(),
+    // Anything but urgent/quiet (absent, unknown, from an older caller) is normal.
+    level: z
+      .unknown()
+      .optional()
+      .transform((v): PushLevel => (v === 'urgent' || v === 'quiet' ? v : 'normal')),
+    // Opaque hash of the session; the relay never sees the session name itself.
+    threadKey: z
+      .string()
+      .regex(/^[0-9a-zA-Z]{1,64}$/)
+      .optional()
+      .catch(undefined),
   })
   .refine((v) => (v.ciphertext === undefined) !== (v.body === undefined), {
     message: 'provide exactly one of body (cleartext) or ciphertext (encrypted)',
@@ -26,15 +39,22 @@ export interface ApnsPayload {
 // schema bounds it well under that, so this is a guard, not a limit callers hit.
 export const APNS_MAX_PAYLOAD_BYTES = 4096;
 
-export function buildApnsPayload(req: PushRequest): ApnsPayload {
+const LEVEL_APS: Record<PushLevel, Record<string, unknown>> = {
+  urgent: { 'interruption-level': 'time-sensitive', 'relevance-score': 1, sound: 'default' },
+  normal: { 'interruption-level': 'active', sound: 'default' },
+  quiet: { 'interruption-level': 'passive', 'relevance-score': 0 },
+};
+
+export function buildApnsPayload(req: Omit<PushRequest, 'level'> & { level?: PushLevel }): ApnsPayload {
+  const tier = { ...LEVEL_APS[req.level ?? 'normal'], ...(req.threadKey ? { 'thread-id': req.threadKey } : {}) };
   if (req.ciphertext !== undefined) {
     return {
       aps: {
         // The NSE replaces this before the user ever sees it. It is only what
         // shows if decryption fails, so it must reveal nothing.
         alert: { title: 'Tether', body: 'New activity' },
         'mutable-content': 1,
-        sound: 'default',
+        ...tier,
       },
       e: req.ciphertext,
       ...(req.deepLink ? { link: req.deepLink } : {}),
@@ -43,7 +63,7 @@ export function buildApnsPayload(req: PushRequest): ApnsPayload {
   return {
     aps: {
       alert: { title: req.title ?? 'Tether', body: req.body },
-      sound: 'default',
+      ...tier,
     },
     ...(req.deepLink ? { link: req.deepLink } : {}),
   };

apps/tether-notify/crypto.go

@@ -30,6 +30,11 @@ type PushContent struct {
 	// Options are a question's labels, for the phone's buttons; only for one
 	// single-choice question.
 	Options []string `json:"options,omitempty"`
+	// Session is the zmx session name; it stays inside the ciphertext, so the relay
+	// never reads it. The phone shows it as the subtitle and groups by it.
+	Session string `json:"session,omitempty"`
+	// Level is the push's urgency (urgent|normal|quiet); see pushLevel.
+	Level string `json:"level,omitempty"`
 }
 
 func encryptPushContent(keyBase64 string, content PushContent) (string, error) {

apps/tether-notify/hold.go

@@ -136,12 +136,13 @@ func runHold(args []string, d holdDeps) error {
 	content := PushContent{
 		Title: project + " · needs you", Body: *body, Link: link,
 		Category: agentCategory(stateWaiting), State: stateWaiting, Version: stored.Version,
+		Session: *session, Level: levelUrgent,
 	}
 	if *kind == "question" {
 		content.Category = questionCategory
 		content.Options = optionButtons(questions)
 	}
-	if err := d.push(content, "agent-"+*session); err != nil {
+	if err := d.push(content, agentCollapseID(*session)); err != nil {
 		fmt.Fprintf(d.stderr, "tether-notify: push for %s failed: %v\n", *session, err)
 		if question {
 			fmt.Fprintln(d.stdout, stored.Version)

apps/tether-notify/hold_test.go

@@ -65,8 +65,9 @@ func TestHoldRecordsAndPushesWhenNobodyIsAttached(t *testing.T) {
 		Title: "project · needs you", Body: "Allow Bash: npm test?",
 		Link:     "tether://session/work?host=devbox",
 		Category: "tether.agent.waiting", State: stateWaiting, Version: s.Version,
+		Session: "work", Level: "urgent",
 	}
-	if !reflect.DeepEqual(p.content, want) || p.collapse != "agent-work" {
+	if !reflect.DeepEqual(p.content, want) || p.collapse != agentCollapseID("work") {
 		t.Fatalf("push %+v %q", p.content, p.collapse)
 	}
 }

apps/tether-notify/level_test.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+func TestPushLevelTiersByCategory(t *testing.T) {
+	for category, want := range map[string]string{
+		"tether.agent.waiting":  levelUrgent,
+		"tether.agent.question": levelUrgent,
+		"tether.agent.done":     levelQuiet,
+		"":                      levelNormal,
+		"something.else":        levelNormal,
+	} {
+		if got := pushLevel(category); got != want {
+			t.Errorf("pushLevel(%q) = %q, want %q", category, got, want)
+		}
+	}
+}
+
+func TestThreadKeyIsStableOpaqueAndPerSession(t *testing.T) {
+	a := threadKey("host", "alpha")
+	if a != threadKey("host", "alpha") {
+		t.Fatal("not stable")
+	}
+	if a == threadKey("host", "beta") || a == threadKey("other", "alpha") {
+		t.Fatal("collides across session or host")
+	}
+	if strings.Contains(a, "alpha") || len(a) != 16 {
+		t.Fatalf("key %q leaks or has wrong length", a)
+	}
+	if threadKey("host", "") != "" {
+		t.Fatal("no session must mean no key")
+	}
+}
+
+func TestRelayRequestCarriesOnlyLevelAndOpaqueKeyInCleartext(t *testing.T) {
+	out, _ := json.Marshal(relayRequest{Token: "t", Ciphertext: "c", CollapseID: "x", Level: levelUrgent, ThreadKey: threadKey("h", "secret-session")})
+	if strings.Contains(string(out), "secret-session") || !strings.Contains(string(out), `"level":"urgent"`) {
+		t.Fatalf("body %s", out)
+	}
+	old, _ := json.Marshal(relayRequest{Token: "t", Ciphertext: "c", CollapseID: "x"})
+	if strings.Contains(string(old), "level") || strings.Contains(string(old), "threadKey") {
+		t.Fatalf("empty fields must be omitted: %s", old)
+	}
+}
+
+func TestAgentCollapseIDHidesTheSessionAndSeparatesSessions(t *testing.T) {
+	work := agentCollapseID("work")
+	if strings.Contains(work, "work") || work != "agent-"+threadKey(hostLabel(), "work") {
+		t.Fatalf("collapse id %q must be the hashed form", work)
+	}
+	if work == agentCollapseID("other") {
+		t.Fatal("two sessions share a collapse id")
+	}
+}

apps/tether-notify/main.go

@@ -2,6 +2,8 @@ package main
 
 import (
 	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"flag"
@@ -23,6 +25,44 @@ type relayRequest struct {
 	Token      string `json:"token"`
 	Ciphertext string `json:"ciphertext"`
 	CollapseID string `json:"collapseId"`
+	// Level and ThreadKey are the only new cleartext: an urgency tier and an opaque
+	// hash, never the session name.
+	Level     string `json:"level,omitempty"`
+	ThreadKey string `json:"threadKey,omitempty"`
+}
+
+const (
+	levelUrgent = "urgent"
+	levelNormal = "normal"
+	levelQuiet  = "quiet"
+)
+
+// pushLevel tiers a push by what the user must do: a question or permission wants an
+// answer, a finished turn can wait, anything else keeps the default.
+func pushLevel(category string) string {
+	switch category {
+	case "tether.agent.waiting", questionCategory:
+		return levelUrgent
+	case "tether.agent.done":
+		return levelQuiet
+	}
+	return levelNormal
+}
+
+// threadKey groups one session's pushes on the phone without telling the relay which
+// session it is: a truncated hash of the host label and session name.
+func threadKey(host, session string) string {
+	if session == "" {
+		return ""
+	}
+	sum := sha256.Sum256([]byte(host + "\x00" + session))
+	return hex.EncodeToString(sum[:8])
+}
+
+// agentCollapseID lets a session's newer agent push replace its older one. It reaches the
+// relay in cleartext as apns-collapse-id, so it carries the hash, not the session name.
+func agentCollapseID(session string) string {
+	return "agent-" + threadKey(hostLabel(), session)
 }
 
 func main() {
@@ -154,6 +194,7 @@ func cmdNotify(args []string) error {
 	body := fs.String("body", "", "notification body")
 	link := fs.String("link", "", "tether:// deep link (optional)")
 	category := fs.String("category", "", "iOS notification category (optional)")
+	session := fs.String("session", "", "zmx session name, shown as the subtitle and used to group pushes (optional)")
 	collapse := fs.String("collapse", "tether-notify", "APNs collapse id")
 	dryRun := fs.Bool("dry-run", false, "print requests instead of sending")
 	if err := fs.Parse(args); err != nil {
@@ -162,7 +203,7 @@ func cmdNotify(args []string) error {
 	if *title == "" || *body == "" {
 		return fmt.Errorf("notify requires --title and --body")
 	}
-	return sendPush(PushContent{Title: *title, Body: *body, Link: *link, Category: *category}, *collapse, *dryRun)
+	return sendPush(PushContent{Title: *title, Body: *body, Link: *link, Category: *category, Session: *session}, *collapse, *dryRun)
 }
 
 func sendPush(content PushContent, collapse string, dryRun bool) error {
@@ -173,6 +214,10 @@ func sendPush(content PushContent, collapse string, dryRun bool) error {
 	if len(devices) == 0 {
 		return fmt.Errorf("no registered devices")
 	}
+	if content.Level == "" {
+		content.Level = pushLevel(content.Category)
+	}
+	thread := threadKey(hostLabel(), content.Session)
 	url := strings.TrimRight(relayURL(), "/") + "/push"
 	client := &http.Client{Timeout: 5 * time.Second}
 
@@ -183,7 +228,7 @@ func sendPush(content PushContent, collapse string, dryRun bool) error {
 			fmt.Fprintf(os.Stderr, "encrypt for %s failed: %v\n", shortToken(device.Token), err)
 			continue
 		}
-		req := relayRequest{Token: device.Token, Ciphertext: ciphertext, CollapseID: collapse}
+		req := relayRequest{Token: device.Token, Ciphertext: ciphertext, CollapseID: collapse, Level: content.Level, ThreadKey: thread}
 		if dryRun {
 			out, _ := json.Marshal(req)
 			fmt.Println(string(out))

apps/tether-notify/state.go

@@ -45,7 +45,7 @@ func runState(args []string, d stateDeps) error {
 	title := fs.String("title", "", "push title")
 	body := fs.String("body", "", "push body / status message")
 	link := fs.String("link", "", "tether:// deep link")
-	collapse := fs.String("collapse", "", "APNs collapse id (default agent-<session>)")
+	collapse := fs.String("collapse", "", "APNs collapse id (default: a hashed per-session id)")
 	dryRun := fs.Bool("dry-run", false, "print the push instead of sending it")
 	if err := fs.Parse(args); err != nil {
 		return err
@@ -60,7 +60,7 @@ func runState(args []string, d stateDeps) error {
 		d.push = func(c PushContent, col string) error { return sendPush(c, col, true) }
 	}
 	if *collapse == "" {
-		*collapse = "agent-" + *session
+		*collapse = agentCollapseID(*session)
 	}
 
 	in := SessionState{Session: *session, Agent: *agent, State: *state, Message: *body, Link: *link, Version: newVersion()}
@@ -103,7 +103,7 @@ func runState(args []string, d stateDeps) error {
 	if clients, err := zmxClients(d.run); err == nil && clients[*session] > 0 {
 		return nil
 	}
-	content := PushContent{Title: *title, Body: *body, Link: *link}
+	content := PushContent{Title: *title, Body: *body, Link: *link, Session: *session, Level: pushLevel(agentCategory(*state))}
 	// Actions need a session link to answer and a saved state to check against.
 	if stored != nil && stored.Version != "" && actionableLink(*link, *session) {
 		content.Category = agentCategory(*state)

apps/tether-notify/state_test.go

@@ -65,7 +65,7 @@ func TestStateDonePushesWhenNobodyAttached(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(*pushes) != 1 || (*pushes)[0].collapse != "agent-work" || (*pushes)[0].content.Link != "tether://session/work?host=h" {
+	if len(*pushes) != 1 || (*pushes)[0].collapse != agentCollapseID("work") || (*pushes)[0].content.Link != "tether://session/work?host=h" {
 		t.Fatalf("pushes %+v", *pushes)
 	}
 	s, _ := readSession("work")
@@ -91,6 +91,21 @@ func TestStatePushesCarryTheAgentCategoryAndState(t *testing.T) {
 	}
 }
 
+func TestStatePushesCarryTheSessionAndALevel(t *testing.T) {
+	for state, want := range map[string]string{"waiting": "urgent", "done": "quiet"} {
+		for _, link := range []string{"tether://session/work?host=h", ""} {
+			d, pushes := fakeDeps(t, "name=work\tclients=0\n", nil)
+			if err := runState(args(state, "--title", "t", "--body", "b", "--link", link), d); err != nil {
+				t.Fatal(err)
+			}
+			got := (*pushes)[0].content
+			if got.Session != "work" || got.Level != want {
+				t.Fatalf("%s link %q: session %q level %q", state, link, got.Session, got.Level)
+			}
+		}
+	}
+}
+
 func TestStateWithoutASessionLinkOffersNoActions(t *testing.T) {
 	for _, link := range []string{
 		"", "https://example.com", "tether://session/work", "tether://session/other?host=h",