feat: gracefully drain seller requests on shutdown

CHANGELOG.md

@@ -8,6 +8,8 @@ This project uses selective package publishing. Each release entry lists the pub
 
 ### Added
 
+- Seller shutdown now stops new work and drains active requests, final payment authorizations, and outgoing transport buffers before closing connections. Configure the default 60-second drain budget with `antseed seller start --shutdown-drain-timeout-ms` or the SDK's `shutdownDrainTimeoutMs` option.
+
 - Desktop telemetry's `app_connect` / `app_disconnect` user actions now carry which app was connected, as an `app` property drawn from a fixed local taxonomy (the packaged profile names plus the Telegram bot); user-added custom apps report as `custom`, so raw app names never leave the device. Connect events are also attributed to the specific app being connected rather than firing on every profile-set restart (profile switches and custom-app removals no longer emit spurious `app_connect`).
 - Desktop now finds T3 Code installed under any release channel — the launch-target lookup previously only checked for "T3 Code (Alpha)", so stable/Beta/Nightly installs got no app icon, no default "Open with" application, and no restart action. All channel variants are now probed (stable first), and the T3 Code rows fall back to the official t3.codes icon instead of the generic mark when the app isn't installed.
 - Desktop's Home screen keeps the "Use AntSeed on your favorite app" pills visible after connecting a tool — previously connecting anything hid the whole list. The pitch now disappears only once the user has chats, and an already-connected app's pill shows as connected (green dot, green-tinted border) and opens the Apps page instead of reconnecting.

apps/cli/README.md

@@ -272,6 +272,22 @@ antseed buyer start --disable-metadata-v2-services
 
 For production sellers, prefer a dedicated Base JSON-RPC endpoint over public defaults. You can set it durably with `payments.crypto.rpcUrl`, at runtime with `ANTSEED_BASE_RPC_URL`, or for one run with `antseed seller start --base-rpc-url <url>`.
 
+### Graceful seller shutdown
+
+On SIGINT or SIGTERM, the seller stops advertising and rejects new requests with
+503 while allowing running requests, final payment authorizations, and outgoing
+transport buffers to drain. The default drain budget is 60 seconds:
+
+```bash
+antseed seller start --shutdown-drain-timeout-ms 120000
+```
+
+The SDK equivalent is `new AntseedNode({ role: 'seller', shutdownDrainTimeoutMs: 120000 })`.
+Set the timeout to `0` to skip waiting. Requests still running at the deadline are
+disconnected; forced termination (such as SIGKILL) cannot be drained. A service
+manager's termination timeout should exceed the drain budget and leave time for
+settlement and cleanup.
+
 ### Metadata v12 rollout
 
 This release announces metadata v12. Buyers supporting only older metadata versions drop v12 sellers from discovery, while updated buyers continue accepting older v10/v11 sellers. Upgrade buyer CLIs and desktop apps before upgrading sellers. Removing capability or unit-billing fields does not downgrade the metadata version; rollback requires running the older seller binary.

apps/cli/src/cli/commands/seller/start.ts

@@ -377,6 +377,11 @@ export function registerSellerStartCommand(sellerCmd: Command): void {
     .option('-r, --reserve <number>', 'runtime-only reserve floor override (does not write config file)', parseFloat)
     .option('--input-usd-per-million <number>', 'runtime-only input pricing override in USD per 1M tokens', parseFloat)
     .option('--output-usd-per-million <number>', 'runtime-only output pricing override in USD per 1M tokens', parseFloat)
+    .option('--shutdown-drain-timeout-ms <number>', 'maximum wait for active requests and final payment authorizations on shutdown', (value: string) => {
+      const timeout = Number(value)
+      if (!Number.isSafeInteger(timeout) || timeout < 0 || timeout > 2_147_483_647 || value.trim() === '') throw new Error('Shutdown drain timeout must be an integer between 0 and 2147483647')
+      return timeout
+    }, 60_000)
     .option('--dht-port <number>', 'UDP port for DHT (default: 6881)', parseInt)
     .option('--signaling-port <number>', 'TCP port for P2P signaling (default: 6882)', parseInt)
     .option('--min-settle-delta <usdc>', 'minimum unsettled delta (USDC decimal, e.g. 0.002) before idle settle submits a tx')
@@ -692,6 +697,7 @@ export function registerSellerStartCommand(sellerCmd: Command): void {
 
       const node = new AntseedNode({
         role: 'seller',
+        shutdownDrainTimeoutMs: options.shutdownDrainTimeoutMs as number,
         displayName: config.identity.displayName,
         ...(config.seller.publicAddress ? { publicAddress: config.seller.publicAddress } : {}),
         ...(effectiveSellerConfig.verifications ? { verifications: effectiveSellerConfig.verifications } : {}),

packages/node/src/discovery/announcer.ts

@@ -123,6 +123,7 @@ export class PeerAnnouncer {
 
   async announce(): Promise<void> {
     this._latestMetadata = await this._buildSignedMetadata(true);
+    if (this.stopped) return;
 
     const failures = await this._announceTopics();
     if (failures > 0) {

packages/node/src/node.ts

@@ -221,6 +221,7 @@ export interface NodeVerificationConfig {
 
 export interface NodeConfig {
   role: 'seller' | 'buyer';
+  shutdownDrainTimeoutMs?: number;
   displayName?: string;
   /** Publicly reachable seller address override ("host:port") announced in metadata. */
   publicAddress?: string;
@@ -330,6 +331,7 @@ export class AntseedNode extends EventEmitter {
   private _provers: Prover[] = [];
   private _router: Router | null = null;
   private _started = false;
+  private _stopPromise: Promise<void> | null = null;
   private _announcer: PeerAnnouncer | null = null;
   /** Set while advertising is paused (e.g. seller wallet out of gas). */
   private _advertisingPausedReason: string | null = null;
@@ -394,6 +396,9 @@ export class AntseedNode extends EventEmitter {
 
   constructor(config: NodeConfig) {
     super();
+    if (config.shutdownDrainTimeoutMs !== undefined && (!Number.isSafeInteger(config.shutdownDrainTimeoutMs) || config.shutdownDrainTimeoutMs < 0 || config.shutdownDrainTimeoutMs > 2_147_483_647)) {
+      throw new Error('shutdownDrainTimeoutMs must be an integer between 0 and 2147483647');
+    }
     this._config = config;
   }
 
@@ -434,6 +439,7 @@ export class AntseedNode extends EventEmitter {
 
   /** Resume DHT announcements after `pauseAdvertising` (announces immediately). */
   resumeAdvertising(): void {
+    if (this._stopPromise) return;
     if (this._advertisingPausedReason === null) return;
     this._advertisingPausedReason = null;
     this._announcer?.startPeriodicAnnounce();
@@ -573,11 +579,44 @@ export class AntseedNode extends EventEmitter {
     this.emit("started");
   }
 
-  async stop(): Promise<void> {
+  stop(): Promise<void> {
+    if (!this._started) return Promise.resolve();
+    if (!this._stopPromise) {
+      this._stopPromise = this._stop().finally(() => { this._stopPromise = null; });
+    }
+    return this._stopPromise;
+  }
+
+  private async _stop(): Promise<void> {
     if (!this._started) {
       return;
     }
 
+    if (this._sellerHandler) {
+      const timeoutMs = this._config.shutdownDrainTimeoutMs ?? 60_000;
+      const deadline = Date.now() + timeoutMs;
+      this._sellerPaymentManager?.beginDrain();
+      const draining = this._sellerHandler.drain(timeoutMs);
+      void this.pauseAdvertising('shutting-down').catch((err: unknown) => {
+        debugWarn(`[Node] Could not refresh shutdown metadata: ${String(err)}`);
+      });
+      const completed = await draining;
+      if (!completed) {
+        debugWarn(`[Node] Seller drain timed out after ${timeoutMs}ms; closing remaining requests`);
+      } else {
+        await this._sellerPaymentManager?.drainPendingPayments(Math.max(0, deadline - Date.now()));
+      }
+      await Promise.all([...this._connectionManager?.connections.values() ?? []].map(async (connection) => {
+        try {
+          if (!await connection.drainOutgoing(Math.max(0, deadline - Date.now()))) {
+            debugWarn('[Node] Shutdown deadline reached before outgoing transport buffers drained');
+          }
+        } catch (err) {
+          debugWarn(`[Node] Could not drain outgoing transport: ${String(err)}`);
+        }
+      }));
+    }
+
     // Give in-transit NeedAuth messages time to arrive on the DataChannel,
     // then wait for their handlers to finish. This ensures the seller has a
     // valid SpendingAuth for settlement before we close the connection.

packages/node/src/p2p/connection-manager.ts

@@ -295,6 +295,19 @@ export class PeerConnection extends EventEmitter {
     }
   }
 
+  async drainOutgoing(timeoutMs: number): Promise<boolean> {
+    const deadline = Date.now() + timeoutMs;
+    while (true) {
+      const buffered = this._dataChannel?.isOpen()
+        ? this._dataChannel.bufferedAmount()
+        : this._rawSocket?.writableLength ?? 0;
+      if (buffered === 0) return true;
+      const remaining = deadline - Date.now();
+      if (remaining <= 0) return false;
+      await new Promise<void>((resolve) => setTimeout(resolve, Math.min(10, remaining)));
+    }
+  }
+
   /** Send a message through the active transport. */
   send(data: Uint8Array): void {
     if (this._state !== ConnectionState.Open && this._state !== ConnectionState.Authenticated) {

packages/node/src/payments/seller-payment-manager.ts

@@ -87,6 +87,7 @@ interface LatestAuth {
  * The seller tracks spending locally and settles/closes via the contract at session end.
  */
 export class SellerPaymentManager {
+  private _draining = false;
   private readonly _signer: AbstractSigner;
   private readonly _channelsClient: ChannelsClient;
   private readonly _config: SellerPaymentConfig;
@@ -501,6 +502,7 @@ export class SellerPaymentManager {
       const channelsDomain = makeChannelsDomain(this._config.chainId, channelsAddr);
 
       if (existingCumulative === undefined) {
+        if (this._draining) return 'rejected';
         const hasReserveFields = payload.reserveSalt != null
           || payload.reserveMaxAmount != null
           || payload.reserveDeadline != null;
@@ -1524,6 +1526,32 @@ export class SellerPaymentManager {
     };
   }
 
+  async drainPendingPayments(timeoutMs: number): Promise<void> {
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    try {
+      const complete = await Promise.race([
+        (async () => {
+          const deadline = Date.now() + timeoutMs;
+          await Promise.all([...this._buyerLocks.values()]);
+          const results = await Promise.all(this._channelStore.getActiveChannels(CHANNEL_ROLE.SELLER).map(async (channel) => {
+            const reached = await this.awaitAcceptedAtLeast(channel.sessionId, this.getCumulativeSpend(channel.sessionId), Math.max(0, deadline - Date.now()));
+            await this.waitForPendingAuths(channel.peerId);
+            return reached;
+          }));
+          return results.every(Boolean);
+        })(),
+        new Promise<false>((resolve) => { timer = setTimeout(() => resolve(false), timeoutMs); }),
+      ]);
+      if (!complete) debugWarn('[SellerPayment] Shutdown deadline reached before all final spending authorizations arrived');
+    } finally {
+      clearTimeout(timer);
+    }
+  }
+
+  beginDrain(): void {
+    this._draining = true;
+  }
+
   // ── Buyer-requested cooperative close ─────────────────────────
 
   /**

packages/node/src/seller-request-handler.ts

@@ -84,11 +84,29 @@ export class SellerRequestHandler {
   private readonly _providerLoadCounts = new Map<string, number>();
   private readonly _attestRateWindows = new Map<string, { start: number; count: number }>();
   private _metadataRefreshTimer: ReturnType<typeof setTimeout> | null = null;
+  private _draining = false;
+  private _aborted = false;
+  private readonly _pendingRequests = new Set<Promise<void>>();
 
   constructor(deps: SellerRequestHandlerDeps) {
     this._deps = deps;
   }
 
+  async drain(timeoutMs: number): Promise<boolean> {
+    this._draining = true;
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    try {
+      const completed = await Promise.race([
+        Promise.allSettled([...this._pendingRequests]).then(() => true),
+        new Promise<false>((resolve) => { timer = setTimeout(() => resolve(false), timeoutMs); }),
+      ]);
+      this._aborted = !completed;
+      return completed;
+    } finally {
+      clearTimeout(timer);
+    }
+  }
+
   private _allowAttest(buyerPeerId: string): boolean {
     const now = Date.now();
     const win = this._attestRateWindows.get(buyerPeerId);
@@ -123,7 +141,7 @@ export class SellerRequestHandler {
       maxUploadBodyBytes: this._deps.maxUploadBodyBytes,
     });
 
-    mux.onProxyRequest(async (request: SerializedHttpRequest) => {
+    const processRequest = async (request: SerializedHttpRequest): Promise<void> => {
       debugLog(`[SellerHandler] Received request: ${request.method} ${request.path} (reqId=${request.requestId.slice(0, 8)})`);
 
       // Handle /v1/models locally — free metadata endpoint, no payment required.
@@ -183,13 +201,15 @@ export class SellerRequestHandler {
             headers: request.headers,
             body: request.body,
           });
+          if (this._aborted) return;
           mux.sendProxyResponse({
             requestId: request.requestId,
             statusCode: resp.statusCode,
             headers: resp.headers,
             body: resp.body,
           });
         } catch (err) {
+          if (this._aborted) return;
           const message = err instanceof Error ? err.message : String(err);
           mux.sendProxyResponse({
             requestId: request.requestId,
@@ -293,6 +313,7 @@ export class SellerRequestHandler {
           // that has queued later auths behind its per-buyer mutex) so we don't
           // 402 against a stale accepted cumulative.
           await spm.waitForPendingAuths(buyerPeerId);
+          if (this._aborted) return;
           // Re-read after the await — the session may have been evicted (timeout
           // checker, disconnect) while the on-chain top-up was confirming.
           const session = spm.getChannelByPeer(buyerPeerId);
@@ -458,6 +479,7 @@ export class SellerRequestHandler {
         ...request,
         headers: { ...request.headers },
       };
+      if (this._aborted) return;
 
       // Track active seller session at request start
       this._deps.sessionTracker?.getOrCreateSession(buyerPeerId, provider.name);
@@ -500,6 +522,7 @@ export class SellerRequestHandler {
         try {
           const response = await this._executeRequest(provider, request, {
             onResponseStart: (streamResponseStart) => {
+              if (this._aborted) return;
               streamedResponseStarted = true;
               responseStartedAt = Date.now();
               statusCode = streamResponseStart.statusCode;
@@ -508,6 +531,7 @@ export class SellerRequestHandler {
               mux.sendProxyResponse(streamResponseStart);
             },
             onResponseChunk: (chunk) => {
+              if (this._aborted) return;
               if (!streamedResponseStarted) return;
               // Hold the done chunk — send it after usage is parsed so we can append cost trailer
               if (chunk.done) {
@@ -517,6 +541,7 @@ export class SellerRequestHandler {
               mux.sendProxyChunk(chunk);
             },
           });
+          if (this._aborted) return;
           statusCode = response.statusCode;
           responseBody = response.body ?? new Uint8Array(0);
           responseForAuth = response;
@@ -547,6 +572,7 @@ export class SellerRequestHandler {
             });
           }
         } catch (err) {
+          if (this._aborted) return;
           const message = err instanceof Error ? err.message : "Internal error";
           debugWarn(`[SellerHandler] Provider exception: provider="${provider.name}" model="${requestedModel}" buyer=${buyerPeerId.slice(0, 12)}... (${Date.now() - startTime}ms) ${message}`);
           responseBody = new TextEncoder().encode(message);
@@ -611,6 +637,7 @@ export class SellerRequestHandler {
             providerUsage: responseUsage,
           });
         }
+        if (this._aborted) return;
 
         // Record spend and send NeedAuth with cost data after every request.
         // The buyer validates the cost independently and responds with SpendingAuth.
@@ -678,6 +705,25 @@ export class SellerRequestHandler {
         this.adjustProviderLoad(provider.name, -1);
         if (isBillable) spm!.endBillableRequest(buyerPeerId);
       }
+    };
+
+    mux.onProxyRequest(async (request: SerializedHttpRequest) => {
+      if (this._draining) {
+        mux.sendProxyResponse({
+          requestId: request.requestId,
+          statusCode: 503,
+          headers: { 'content-type': 'application/json', 'retry-after': '2' },
+          body: new TextEncoder().encode(JSON.stringify({ error: 'seller_shutting_down' })),
+        });
+        return;
+      }
+      const operation = processRequest(request);
+      this._pendingRequests.add(operation);
+      try {
+        await operation;
+      } finally {
+        this._pendingRequests.delete(operation);
+      }
     });
 
     return { mux };
@@ -974,7 +1020,7 @@ export class SellerRequestHandler {
   }
 
   private _scheduleMetadataRefresh(): void {
-    if (!this._deps.announcer || this._metadataRefreshTimer) {
+    if (this._draining || !this._deps.announcer || this._metadataRefreshTimer) {
       return;
     }