refactor: split big files

docs/qa/journeys/J-operate-daemon-schema.md

@@ -10,7 +10,9 @@ flowchart TD
     DB -->|fresh or current| BOOT[Daemon applies global + memory streams]
     DB -->|pre-Goose table found| REFUSE[Startup refuses before readiness and names remediation]
     DB -->|recorded version ahead| AHEAD[Startup refuses and requests a newer binary or isolated fresh home]
-    REFUSE --> CHOICE{Preserve old alpha state?}
+    DB -->|SQLite corruption| CORRUPT[Startup refuses and leaves DB, WAL, and SHM unchanged]
+    REFUSE --> CHOICE{Preserve refused state?}
+    CORRUPT --> CHOICE
     CHOICE -->|yes| BACKUP[Stop AGH and move the complete AGH_HOME or workspace .agh family]
     CHOICE -.->|not ready to discard| ABANDON[Abandon: keep old home stopped for later inspection]
     REFUSE --> DIRECT[Confirm stopped-daemon extension/MCP/provider-auth direct opens return the same typed refusal]
@@ -33,7 +35,7 @@ flowchart TD
 journey:
   id: J-operate-daemon-schema
   name: "Start and inspect the daemon schema safely"
-  value_statement: "An operator can start AGH without silently rewriting incompatible alpha state and can inspect the exact daemon-global schema versions through agent-manageable surfaces."
+  value_statement: "An operator can start AGH without silently rewriting incompatible or corrupt state and can inspect the exact daemon-global schema versions through agent-manageable surfaces."
   personas: [Bruno, Ada]
   entry_points:
     - url: "CLI: agh daemon start"
@@ -53,13 +55,13 @@ journey:
   actions:
     - step: 1
       verb: "Start the daemon against the selected AGH_HOME"
-      expected_observable: "A current database reaches readiness; a pre-Goose database is refused before mutation with its path and remediation."
+      expected_observable: "A current database reaches readiness; a pre-Goose or corrupt database is refused before mutation with its path."
     - step: 2
       verb: "Stop AGH, preserve or move the complete incompatible AGH_HOME or workspace .agh family, and select a separate fresh home"
       expected_observable: "Every sibling database remains together for investigation and the separately selected fresh daemon home reaches readiness."
     - step: 3
       verb: "Read preserved session events or materialize a terminal ledger"
-      expected_observable: "Legacy and ahead events.db files are refused before domain queries while a current session database remains readable."
+      expected_observable: "Legacy, ahead, and corrupt events.db files are refused before domain queries while a current session database remains readable."
     - step: 4
       verb: "Exercise global and memory read paths after migration and restart"
       expected_observable: "Workspace and memory catalog reads succeed while the two version streams remain independently visible."
@@ -82,7 +84,7 @@ journey:
   crosses: [daemon-boot, session-readers, ledger-materialization, SQLite, Goose, HTTP, UDS, CLI]
 ```
 
-Taxonomy note: this journey covers the functional happy path, global and per-session legacy/ahead error and
+Taxonomy note: this journey covers the functional happy path, global and per-session legacy/ahead/corruption error and
 recovery paths, shared-file domain smoke, structured-surface consistency, and operational recoverability. UI
 responsiveness, mobile continuity, and screen-reader coverage are deliberately skipped because the program adds
 no rendered UI.

docs/qa/scenarios/RT-preserve-corrupt-database-family.md

@@ -0,0 +1,21 @@
+---
+id: RT-preserve-corrupt-database-family
+area: RT
+title: Preserve a corrupt database family
+persona: Bruno
+journey: J-operate-daemon-schema
+expected: Startup or a session read refuses the corrupt database with its path while the database, WAL, and SHM bytes remain identical and no quarantine or replacement file appears.
+entry_points: agh daemon start --foreground; agh session events <session-id>
+qa_status: untested
+bug_ids:
+fix_status:
+retest_status:
+fix_commits:
+evidence:
+last_report:
+overlaps: RT-refuse-legacy-database;RT-refuse-ahead-database
+---
+
+Flagged after the store opener stopped performing context-free automatic corruption recovery. The next targeted
+cycle should hash the complete database family before and after each refusal and inspect the filesystem for
+unexpected `.corrupt.*` files or fresh replacement databases. No QA session ran as part of this change.

extensions/bridges/discord/discord_configuration.go

@@ -0,0 +1,300 @@
+package main
+
+import (
+	"context"
+
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"os"
+
+	"strings"
+
+	"time"
+
+	bridgepkg "github.com/compozy/agh/internal/bridges/contract"
+	"github.com/compozy/agh/internal/bridgesdk"
+	"github.com/compozy/agh/internal/subprocess"
+)
+
+func (p *discordProvider) stopResources() {
+	progressToClose := p.takeDiscordProgressDispatchers()
+	batchersToClose := make(map[*bridgesdk.InboundBatcher]struct{})
+	routes := p.routes.Snapshot()
+	for instanceID := range routes {
+		cfg := routes[instanceID]
+		if cfg.batcher == nil {
+			continue
+		}
+		batchersToClose[cfg.batcher] = struct{}{}
+		p.routes.Update(instanceID, func(current resolvedInstanceConfig) resolvedInstanceConfig {
+			current.batcher = nil
+			return current
+		})
+	}
+	closeDiscordInboundBatchers(batchersToClose)
+	closeDiscordProgressDispatchers(progressToClose)
+}
+
+func (p *discordProvider) reconcileInstanceConfigs(
+	ctx context.Context,
+	session *bridgesdk.Session,
+	managed []subprocess.InitializeBridgeManagedInstance,
+) []resolvedInstanceConfig {
+	batchersToClose := make(map[*bridgesdk.InboundBatcher]struct{})
+	reconciler := bridgesdk.ManagedConfigReconciler[resolvedInstanceConfig]{
+		Routes:   p.routes,
+		Resolve:  p.resolveInstanceConfig,
+		Prepare:  p.prepareDiscordConfigs,
+		Finalize: p.finalizeDiscordConfigs,
+		Identity: func(config resolvedInstanceConfig) string { return config.instanceID },
+		Merge: func(prior resolvedInstanceConfig, next resolvedInstanceConfig) resolvedInstanceConfig {
+			if prior.batcher != nil && prior.batcher != next.batcher {
+				batchersToClose[prior.batcher] = struct{}{}
+			}
+			return next
+		},
+		OnRemoved: func(config resolvedInstanceConfig) error {
+			if config.batcher != nil {
+				batchersToClose[config.batcher] = struct{}{}
+			}
+			return nil
+		},
+		OnPublish: func() { closeDiscordInboundBatchers(batchersToClose) },
+	}
+	configs, err := reconciler.Reconcile(ctx, session, managed)
+	if err != nil {
+		p.setLastError(err)
+		return nil
+	}
+	return configs
+}
+
+func (p *discordProvider) finalizeDiscordConfigs(
+	ctx context.Context,
+	_ *bridgesdk.Session,
+	configs []resolvedInstanceConfig,
+) ([]resolvedInstanceConfig, error) {
+	for idx := range configs {
+		status, degradation, probeErr := p.determineInitialState(ctx, &configs[idx])
+		if probeErr != nil {
+			p.setLastError(probeErr)
+		}
+		configs[idx].initialStatus = status
+		configs[idx].initialDegradation = degradation
+	}
+	return configs, nil
+}
+
+func (p *discordProvider) prepareDiscordConfigs(
+	_ context.Context,
+	_ *bridgesdk.Session,
+	configs []resolvedInstanceConfig,
+) ([]resolvedInstanceConfig, error) {
+	if len(configs) == 0 {
+		return configs, nil
+	}
+	requestedListen := strings.TrimSpace(os.Getenv(discordListenAddrEnv))
+	usedPaths := make(map[string]string, len(configs))
+
+	for idx := range configs {
+		requestedListen = applyDiscordListenConstraint(&configs[idx], requestedListen)
+		applyDiscordWebhookPathConflict(&configs[idx], usedPaths)
+	}
+	p.applyDiscordListenErrors(configs, requestedListen)
+	p.mu.Lock()
+	p.listenAddr = requestedListen
+	p.mu.Unlock()
+	return configs, nil
+}
+
+func applyDiscordListenConstraint(cfg *resolvedInstanceConfig, requestedListen string) string {
+	if cfg == nil || cfg.listenAddr == "" {
+		return requestedListen
+	}
+	if requestedListen == "" {
+		return cfg.listenAddr
+	}
+	if requestedListen != cfg.listenAddr && cfg.configError == nil {
+		cfg.configError = fmt.Errorf(
+			"discord: instance %q configured incompatible listen_addr %q (runtime uses %q)",
+			cfg.instanceID,
+			cfg.listenAddr,
+			requestedListen,
+		)
+	}
+	return requestedListen
+}
+
+func applyDiscordWebhookPathConflict(cfg *resolvedInstanceConfig, usedPaths map[string]string) {
+	if cfg == nil || cfg.webhookPath == "" {
+		return
+	}
+	if owner, ok := usedPaths[cfg.webhookPath]; ok && cfg.configError == nil {
+		cfg.configError = fmt.Errorf(
+			"discord: webhook path %q is shared by %q and %q",
+			cfg.webhookPath,
+			owner,
+			cfg.instanceID,
+		)
+	}
+	usedPaths[cfg.webhookPath] = cfg.instanceID
+}
+
+func (p *discordProvider) applyDiscordListenErrors(configs []resolvedInstanceConfig, requestedListen string) {
+	if requestedListen == "" {
+		for idx := range configs {
+			if configs[idx].configError == nil {
+				configs[idx].configError = errors.New("discord: webhook listen address is required")
+			}
+		}
+		return
+	}
+	if err := p.startServer(requestedListen); err != nil {
+		for idx := range configs {
+			if configs[idx].configError == nil {
+				configs[idx].configError = err
+			}
+		}
+	}
+}
+
+func (p *discordProvider) resolveInstanceConfig(
+	session *bridgesdk.Session,
+	managed subprocess.InitializeBridgeManagedInstance,
+) resolvedInstanceConfig {
+	cfg := discordProviderConfig{}
+	if len(managed.Instance.ProviderConfig) > 0 {
+		if err := json.Unmarshal(managed.Instance.ProviderConfig, &cfg); err != nil {
+			return resolvedInstanceConfig{
+				managed:     managed,
+				instanceID:  managed.Instance.ID,
+				configError: fmt.Errorf("discord: decode provider_config for %q: %w", managed.Instance.ID, err),
+			}
+		}
+	}
+
+	botToken, _ := session.Cache().BoundSecretValue(managed.Instance.ID, "bot_token")
+	publicKey, _ := session.Cache().BoundSecretValue(managed.Instance.ID, "public_key")
+	listenAddr := firstNonEmpty(cfg.Webhook.ListenAddr, strings.TrimSpace(os.Getenv(discordListenAddrEnv)))
+	webhookPath := normalizeWebhookPath(
+		firstNonEmpty(cfg.Webhook.Path, "/discord/"+strings.TrimSpace(managed.Instance.ID)),
+	)
+	apiBaseURL := normalizeURL(firstNonEmpty(
+		strings.TrimSpace(os.Getenv(discordAPIBaseEnv)),
+		discordDefaultAPIBaseURL,
+	))
+
+	resolved := resolvedInstanceConfig{
+		managed:         managed,
+		instanceID:      strings.TrimSpace(managed.Instance.ID),
+		listenAddr:      listenAddr,
+		webhookPath:     webhookPath,
+		apiBaseURL:      apiBaseURL,
+		applicationID:   strings.TrimSpace(cfg.ApplicationID),
+		botToken:        strings.TrimSpace(botToken),
+		publicKey:       strings.TrimSpace(publicKey),
+		dmPolicy:        managed.Instance.DMPolicy.Normalize(),
+		allowUserIDs:    buildDiscordIDSet(cfg.DM.AllowUserIDs),
+		allowUsernames:  buildDiscordUsernameSet(cfg.DM.AllowUsernames),
+		pairedUserIDs:   buildDiscordIDSet(cfg.DM.PairedUserIDs),
+		pairedUsernames: buildDiscordUsernameSet(cfg.DM.PairedUsernames),
+		dedup:           bridgesdk.NewDedupCache(5*time.Minute, 4000),
+		rateLimiter:     bridgesdk.NewFixedWindowRateLimiter(200, time.Minute),
+		inFlightLimiter: bridgesdk.NewInFlightLimiter(24),
+	}
+	if resolved.dmPolicy == "" {
+		resolved.dmPolicy = bridgepkg.BridgeDMPolicyOpen
+	}
+	if resolved.webhookPath == "" {
+		resolved.configError = errors.New("discord: webhook path is required")
+		return resolved
+	}
+
+	if cfg.Batching.DelayMS > 0 {
+		batcher, err := bridgesdk.NewInboundBatcher(bridgesdk.InboundBatcherConfig{
+			Context: context.Background(),
+			Delay:   time.Duration(cfg.Batching.DelayMS) * time.Millisecond,
+			SplitDelay: func() time.Duration {
+				if cfg.Batching.SplitDelayMS <= 0 {
+					return time.Duration(cfg.Batching.DelayMS) * time.Millisecond
+				}
+				return time.Duration(cfg.Batching.SplitDelayMS) * time.Millisecond
+			}(),
+			SplitThreshold: cfg.Batching.SplitThreshold,
+			Dispatch: func(ctx context.Context, batch bridgesdk.InboundBatch) error {
+				return p.dispatchInboundBatch(ctx, resolved.instanceID, batch)
+			},
+			Now: func() time.Time { return p.now() },
+		})
+		if err != nil {
+			resolved.configError = err
+			return resolved
+		}
+		resolved.batcher = batcher
+	}
+
+	return resolved
+}
+
+func (p *discordProvider) determineInitialState(
+	ctx context.Context,
+	cfg *resolvedInstanceConfig,
+) (bridgepkg.BridgeStatus, *bridgepkg.BridgeDegradation, error) {
+	if cfg == nil {
+		return bridgepkg.BridgeStatusError, nil, errors.New("discord: config is required")
+	}
+	if cfg.configError != nil {
+		return bridgepkg.BridgeStatusDegraded, &bridgepkg.BridgeDegradation{
+			Reason:  bridgepkg.BridgeDegradationReasonTenantConfigInvalid,
+			Message: cfg.configError.Error(),
+		}, cfg.configError
+	}
+	if strings.TrimSpace(cfg.botToken) == "" {
+		err := errors.New("discord: bot_token secret binding is required")
+		return bridgepkg.BridgeStatusAuthRequired, &bridgepkg.BridgeDegradation{
+			Reason:  bridgepkg.BridgeDegradationReasonAuthFailed,
+			Message: err.Error(),
+		}, err
+	}
+	if _, err := decodeDiscordPublicKey(cfg.publicKey); err != nil {
+		wrapped := fmt.Errorf("discord: public_key secret binding invalid: %w", err)
+		return bridgepkg.BridgeStatusAuthRequired, &bridgepkg.BridgeDegradation{
+			Reason:  bridgepkg.BridgeDegradationReasonAuthFailed,
+			Message: wrapped.Error(),
+		}, wrapped
+	}
+	bot, err := p.apiFactory(*cfg).GetBotUser(ctx)
+	if err != nil {
+		classified := bridgesdk.ClassifyError(err)
+		recovery := classified.Recovery()
+		status := recovery.Status
+		if status == "" {
+			status = bridgepkg.BridgeStatusError
+		}
+		if recovery.Degradation != nil {
+			return status, recovery.Degradation, err
+		}
+		return status, &bridgepkg.BridgeDegradation{
+			Reason:  bridgepkg.BridgeDegradationReasonProviderTimeout,
+			Message: classified.Message,
+		}, err
+	}
+	if cfg.applicationID != "" && strings.TrimSpace(bot.ID) != "" && strings.TrimSpace(bot.ID) != cfg.applicationID {
+		err := fmt.Errorf("discord: application_id %q does not match bot identity %q", cfg.applicationID, bot.ID)
+		return bridgepkg.BridgeStatusDegraded, &bridgepkg.BridgeDegradation{
+			Reason:  bridgepkg.BridgeDegradationReasonTenantConfigInvalid,
+			Message: err.Error(),
+		}, err
+	}
+	return bridgepkg.BridgeStatusReady, nil, nil
+}
+
+func (p *discordProvider) startServer(listenAddr string) error {
+	if err := p.http.Start(listenAddr); err != nil {
+		return fmt.Errorf("discord: %w", err)
+	}
+	p.markers.RecordListen(p.http.Address())
+	return nil
+}