diff --git a/src/Usage/Accumulator.php b/src/Usage/Accumulator.php
index b898be9..dac544d 100644
--- a/src/Usage/Accumulator.php
+++ b/src/Usage/Accumulator.php
@@ -16,7 +16,7 @@ class Accumulator
     private Usage $usage;
 
     /**
-     * @var array<string, array{tenant: string, metric: string, value: int, type: string, tags: array<string, mixed>, time?: \DateTime}>
+     * @var array<string, array{tenant: string, metric: string, value: int, type: string, tags: array<string, mixed>, allowNegative: bool, time?: \DateTime}>
      */
     private array $buffer = [];
 
@@ -38,9 +38,16 @@ public function __construct(Usage $usage)
      * Events fold additively; earliest non-null time wins on merge.
      * Gauges use last-write-wins.
      *
+     * Negative values are rejected by default for every metric, so a buggy
+     * negative count/bandwidth is still caught. A caller that emits a genuine
+     * signed delta (e.g. realtime connections `+1`/`-1`) opts in per call with
+     * `$allowNegative = true`. The library stays generic — the decision lives
+     * with the caller, not with any metric-name knowledge here.
+     *
      * @param array<string,mixed> $tags
+     * @param bool $allowNegative Permit a negative value for this metric (default: reject).
      */
-    public function collect(string $tenant, string $metric, int $value, string $type, array $tags = [], ?\DateTime $time = null): self
+    public function collect(string $tenant, string $metric, int $value, string $type, array $tags = [], ?\DateTime $time = null, bool $allowNegative = false): self
     {
         // Compare against '' rather than empty(): the string "0" is a valid
         // tenant/metric id but empty("0") is true in PHP.
@@ -50,7 +57,7 @@ public function collect(string $tenant, string $metric, int $value, string $type
         if ($metric === '') {
             throw new \InvalidArgumentException('Metric name cannot be empty');
         }
-        if ($value < 0) {
+        if ($value < 0 && !$allowNegative) {
             throw new \InvalidArgumentException('Value cannot be negative');
         }
         if ($type !== Usage::TYPE_EVENT && $type !== Usage::TYPE_GAUGE) {
@@ -68,6 +75,15 @@ public function collect(string $tenant, string $metric, int $value, string $type
         if ($type === Usage::TYPE_EVENT && isset($this->buffer[$key])) {
             // earliest time wins on merge
             $this->buffer[$key]['value'] += $value;
+
+            // The opt-in is a property of the folded row, not of whichever
+            // call happened to create it. Folding a signed delta into an entry
+            // opened by a plain positive must not leave the net row looking
+            // unauthorised: it would be rejected at write time, and since a
+            // failed batch keeps its entries buffered, every later flush would
+            // retry the same rejection.
+            $this->buffer[$key]['allowNegative'] = $this->buffer[$key]['allowNegative'] || $allowNegative;
+
             if ($time !== null && (!isset($this->buffer[$key]['time']) || $time < $this->buffer[$key]['time'])) {
                 $this->buffer[$key]['time'] = $time;
             }
@@ -78,6 +94,7 @@ public function collect(string $tenant, string $metric, int $value, string $type
                 'value' => $value,
                 'type' => $type,
                 'tags' => $tags,
+                'allowNegative' => $allowNegative,
             ];
             if ($time !== null) {
                 $entry['time'] = $time;
diff --git a/src/Usage/Adapter.php b/src/Usage/Adapter.php
index 135f5de..9fd6fea 100644
--- a/src/Usage/Adapter.php
+++ b/src/Usage/Adapter.php
@@ -102,6 +102,26 @@ abstract public function purge(string $tenant, array $queries = [], ?string $typ
      */
     abstract public function find(string $tenant, array $queries = [], ?string $type = null): array;
 
+    /**
+     * Find metrics across every tenant in shared-tables mode.
+     *
+     * Deliberately crosses the per-tenant isolation every other read enforces,
+     * so it is reserved for operator-side aggregation jobs that roll many
+     * tenants up in one pass. Never reachable from a tenant-scoped request
+     * path. Pair with `groupBy('tenant')` to keep the rows attributable.
+     *
+     * Adapters that cannot express an unscoped read leave this unsupported.
+     *
+     * @param array<\Utopia\Query\Query> $queries
+     * @param string|null $type Metric type: 'event', 'gauge', or null (query both)
+     * @return array<Metric>
+     * @throws \Exception
+     */
+    public function findAcrossTenants(array $queries = [], ?string $type = null): array
+    {
+        throw new \Exception($this->getName() . ' does not support cross-tenant reads');
+    }
+
     /**
      * Count metrics using Query objects.
      *
diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php
index 068b5bc..37e29be 100644
--- a/src/Usage/Adapter/ClickHouse.php
+++ b/src/Usage/Adapter/ClickHouse.php
@@ -1115,6 +1115,13 @@ private function validateGroupByAttribute(string $attribute, string $type): bool
     {
         $allowed = $type === Usage::TYPE_GAUGE ? Metric::GAUGE_COLUMNS : Metric::EVENT_COLUMNS;
 
+        // `tenant` is a real column in shared-tables mode, not a dimension on
+        // Metric. Grouping by it is what makes a cross-tenant read
+        // ({@see findAcrossTenants}) attributable, so allow it there.
+        if ($this->sharedTables) {
+            $allowed[] = 'tenant';
+        }
+
         if (in_array($attribute, $allowed, true)) {
             return true;
         }
@@ -1279,9 +1286,10 @@ private function getColumnCodec(string $id): string
      * @param string $type Metric type ('event' or 'gauge')
      * @param array<string,mixed> $tags Tags
      * @param int|null $metricIndex Index for batch error messages
+     * @param bool $allowNegative Permit a negative value for this row (default: reject)
      * @throws Exception
      */
-    private function validateMetricData(string $metric, int $value, string $type, array $tags, ?int $metricIndex = null): void
+    private function validateMetricData(string $metric, int $value, string $type, array $tags, ?int $metricIndex = null, bool $allowNegative = false): void
     {
         $prefix = $metricIndex !== null ? "Metric #{$metricIndex}: " : '';
 
@@ -1293,7 +1301,11 @@ private function validateMetricData(string $metric, int $value, string $type, ar
             throw new Exception($prefix . 'Metric exceeds maximum size of 255 characters');
         }
 
-        if ($value < 0) {
+        // Negatives are rejected by default so a buggy negative count/bandwidth
+        // is caught. A row opts in with `allowNegative` for genuine signed
+        // deltas (realtime connections emit +1/-1). The library stays generic —
+        // the caller decides which metrics may be negative.
+        if ($value < 0 && !$allowNegative) {
             throw new Exception($prefix . 'Value cannot be negative');
         }
 
@@ -1331,7 +1343,10 @@ private function validateMetricsBatch(array $metrics, string $type): void
 
             /** @var array<string, mixed> */
             $tags = $metricData['tags'] ?? [];
-            $this->validateMetricData($metric, $value, $type, $tags, $index);
+            // `allowNegative` is a validation-only flag carried on the row; it
+            // gates the negative-value guard and is never stored as a column.
+            $allowNegative = (bool) ($metricData['allowNegative'] ?? false);
+            $this->validateMetricData($metric, $value, $type, $tags, $index, $allowNegative);
 
             $hasTenant = array_key_exists('tenant', $metricData);
 
@@ -1452,6 +1467,44 @@ public function find(string $tenant, array $queries = [], ?string $type = null):
     {
         $this->setOperationContext('find()');
 
+        return $this->findScoped($tenant, $queries, $type);
+    }
+
+    /**
+     * Find metrics across every tenant. Applies no tenant filter, so it is
+     * restricted to shared-tables mode and reserved for operator-side
+     * aggregation jobs — see {@see Adapter::findAcrossTenants()}.
+     *
+     * Callers should add `groupBy('tenant')` to keep rows attributable; the
+     * aggregated paths already carry `tenant` through select/group-by in
+     * shared-tables mode.
+     *
+     * @param array<Query> $queries
+     * @param string|null $type
+     * @return array<Metric>
+     * @throws Exception
+     */
+    public function findAcrossTenants(array $queries = [], ?string $type = null): array
+    {
+        $this->setOperationContext('findAcrossTenants()');
+
+        if (!$this->sharedTables) {
+            throw new Exception('findAcrossTenants() requires shared-tables mode; use find() instead');
+        }
+
+        return $this->findScoped(null, $queries, $type);
+    }
+
+    /**
+     * Shared body for find()/findAcrossTenants(). A null $tenant means no
+     * tenant filter (cross-tenant); a string scopes to that tenant.
+     *
+     * @param array<Query> $queries
+     * @return array<Metric>
+     * @throws Exception
+     */
+    private function findScoped(?string $tenant, array $queries, ?string $type): array
+    {
         if ($type !== null) {
             return $this->findFromTable($tenant, $queries, $type);
         }
@@ -1533,12 +1586,15 @@ private function queriesMatchType(array $queries, string $type): bool
      * - Gauges: SELECT metric, argMax(value, time) as value, toStartOfInterval(time, INTERVAL ...) as time
      * Results are grouped by metric and time bucket, ordered by time ASC.
      *
+     * An `aggregate('max')` query overrides the per-type default value
+     * expression — see {@see findAggregatedFromTable()}.
+     *
      * @param array<Query> $queries
      * @param string $type 'event' or 'gauge'
      * @return array<Metric>
      * @throws Exception
      */
-    private function findFromTable(string $tenant, array $queries, string $type): array
+    private function findFromTable(?string $tenant, array $queries, string $type): array
     {
         $tableName = $this->getTableForType($type);
         $fromTable = $this->buildTableReference($tableName);
@@ -1551,9 +1607,17 @@ private function findFromTable(string $tenant, array $queries, string $type): ar
             throw new Exception('Cursor pagination cannot be combined with groupByInterval');
         }
 
-        // Route through the aggregated path whenever any aggregation
-        // hint is present — time bucketing, dimension breakdown, or both.
-        if (isset($parsed['groupByInterval']) || !empty($parsed['groupBy'])) {
+        // Route through the aggregated path whenever any aggregation hint is
+        // present — time bucketing, dimension breakdown, an explicit
+        // aggregate(), or any combination. An aggregate() on its own still
+        // counts: `aggregate('max')` with no interval and no dimensions is the
+        // flat "highest value over this window" shape, and without this it
+        // would fall through and return raw rows instead.
+        if (
+            isset($parsed['groupByInterval'])
+            || !empty($parsed['groupBy'])
+            || isset($parsed['aggregate'])
+        ) {
             return $this->findAggregatedFromTable($parsed, $fromTable, $type);
         }
 
@@ -1614,7 +1678,7 @@ private function findFromTable(string $tenant, array $queries, string $type): ar
      *          toStartOfInterval(time, INTERVAL 1 HOUR) as time
      *   FROM table WHERE ... GROUP BY metric, time ORDER BY time ASC
      *
-     * @param array{filters: array<string>, params: array<string, mixed>, orderBy?: array<string>, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array<int, string>} $parsed Parsed query data from parseQueries()
+     * @param array{filters: array<string>, params: array<string, mixed>, orderBy?: array<string>, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array<int, string>, aggregate?: string} $parsed Parsed query data from parseQueries()
      * @param string $fromTable Fully qualified table reference
      * @param string $type 'event' or 'gauge'
      * @return array<Metric>
@@ -1624,11 +1688,18 @@ private function findAggregatedFromTable(array $parsed, string $fromTable, strin
     {
         $hasInterval = isset($parsed['groupByInterval']);
 
-        // Choose aggregation function based on metric type
+        // Choose aggregation function based on metric type. `aggregate('max')`
+        // overrides both defaults: for gauges it takes the highest reading in
+        // the bucket rather than the latest one (argMax), which is what rolling
+        // a sampled level series up to a coarser interval needs.
         $valueExpr = $type === Usage::TYPE_GAUGE
             ? 'argMax(value, time) as value'
             : 'SUM(value) as value';
 
+        if (($parsed['aggregate'] ?? null) === 'max') {
+            $valueExpr = 'max(value) as value';
+        }
+
         // Bucket column is only emitted when time bucketing is requested.
         // Without it the result is a flat aggregate per (metric, …dims).
         $bucketSelect = '';
@@ -3470,15 +3541,18 @@ private function buildOrderBySql(array $orderAttributes, bool $flip = false): ar
      * @param string $tenant Tenant scope (shared-tables mode)
      * @param array<Query> $queries
      * @param string $type 'event' or 'gauge' — used for attribute validation
-     * @return array{filters: array<int, string>, params: array<string, mixed>, orderBy?: array<string>, orderAttributes?: array<int, array{attribute: string, direction: string}>, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array<int, string>, cursor?: array<string, mixed>, cursorDirection?: string}
+     * @return array{filters: array<int, string>, params: array<string, mixed>, orderBy?: array<string>, orderAttributes?: array<int, array{attribute: string, direction: string}>, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array<int, string>, aggregate?: string, cursor?: array<string, mixed>, cursorDirection?: string}
      * @throws Exception
      */
-    private function parseQueries(string $tenant, array $queries, string $type = 'event'): array
+    private function parseQueries(?string $tenant, array $queries, string $type = 'event'): array
     {
-        if ($this->sharedTables) {
-            // An empty tenant would compile to `tenant = ''` and silently read
-            // an empty scope. Fail fast instead, like the write side. ("0" is
-            // a valid tenant id, so check for '' specifically.)
+        // A null tenant is the explicit cross-tenant read used by operator-side
+        // aggregation ({@see findAcrossTenants}) — no tenant filter is applied.
+        // It is deliberately distinct from '': an empty string would compile to
+        // `tenant = ''` and silently read an empty scope, so that still fails
+        // fast, like the write side. ("0" is a valid tenant id, so check for ''
+        // specifically.)
+        if ($this->sharedTables && $tenant !== null) {
             if ($tenant === '') {
                 throw new Exception('Tenant cannot be empty in shared-tables mode');
             }
@@ -3493,6 +3567,7 @@ private function parseQueries(string $tenant, array $queries, string $type = 'ev
         $offset = null;
         $groupByInterval = null;
         $groupBy = [];
+        $aggregate = null;
         $cursor = null;
         $cursorDirection = null;
         $paramCounter = 0;
@@ -3739,6 +3814,16 @@ private function parseQueries(string $tenant, array $queries, string $type = 'ev
                         $groupBy[] = $attribute;
                     }
                     break;
+
+                case UsageQuery::TYPE_AGGREGATE:
+                    $aggValue = $values[0] ?? null;
+                    if (!is_string($aggValue) || !in_array($aggValue, UsageQuery::VALID_AGGREGATES, true)) {
+                        throw new Exception(
+                            'Invalid aggregate: expected one of ' . implode(', ', UsageQuery::VALID_AGGREGATES)
+                        );
+                    }
+                    $aggregate = $aggValue;
+                    break;
             }
         }
 
@@ -3768,6 +3853,10 @@ private function parseQueries(string $tenant, array $queries, string $type = 'ev
             $result['groupBy'] = $groupBy;
         }
 
+        if ($aggregate !== null) {
+            $result['aggregate'] = $aggregate;
+        }
+
         if ($cursor !== null && $cursorDirection !== null) {
             $result['cursor'] = $cursor;
             $result['cursorDirection'] = $cursorDirection;
diff --git a/src/Usage/Adapter/Database.php b/src/Usage/Adapter/Database.php
index dae9526..866aeb5 100644
--- a/src/Usage/Adapter/Database.php
+++ b/src/Usage/Adapter/Database.php
@@ -142,7 +142,7 @@ protected function getColumnDefinition(string $id, string $type = 'event'): stri
      * Each metric carries its own `tenant` (shared-tables mode), so a single
      * batch may span multiple tenants.
      *
-     * @param array<array{tenant: string, metric: string, value: int, tags?: array<string,mixed>}> $metrics
+     * @param array<array{tenant: string, metric: string, value: int, tags?: array<string,mixed>, allowNegative?: bool}> $metrics
      * @param string $type Metric type: 'event' or 'gauge'
      * @param int $batchSize
      * @return bool
@@ -157,7 +157,13 @@ public function addBatch(array $metrics, string $type, int $batchSize = 1000): b
                     throw new \InvalidArgumentException("Invalid type '{$type}'. Allowed: event, gauge");
                 }
 
-                if ($metric['value'] < 0) {
+                // Negatives are rejected by default so a buggy count or
+                // bandwidth figure is still caught. A row carrying a genuine
+                // signed delta opts in, the same contract the accumulator sets
+                // and the ClickHouse adapter honours - an adapter that ignored
+                // it would reject the row on every flush, and a failed batch
+                // keeps its entries buffered.
+                if ($metric['value'] < 0 && !($metric['allowNegative'] ?? false)) {
                     throw new \InvalidArgumentException('Value cannot be negative');
                 }
 
diff --git a/src/Usage/Usage.php b/src/Usage/Usage.php
index a5f65bc..c70a96e 100644
--- a/src/Usage/Usage.php
+++ b/src/Usage/Usage.php
@@ -154,6 +154,24 @@ public function find(string $tenant, array $queries = [], ?string $type = null):
         return $this->adapter->find($tenant, $queries, $type);
     }
 
+    /**
+     * Find metrics across every tenant in shared-tables mode.
+     *
+     * Unlike find(), this applies no tenant filter — it is for operator-side
+     * aggregation jobs that need one pass over all tenants instead of N
+     * per-tenant queries. Keep it out of request-scoped code paths, and add
+     * `groupBy('tenant')` so the returned rows stay attributable.
+     *
+     * @param array<\Utopia\Query\Query> $queries
+     * @param string|null $type Metric type: 'event', 'gauge', or null (query both)
+     * @return array<Metric>
+     * @throws \Exception
+     */
+    public function findAcrossTenants(array $queries = [], ?string $type = null): array
+    {
+        return $this->adapter->findAcrossTenants($queries, $type);
+    }
+
     /**
      * Count metrics using Query objects.
      *
diff --git a/src/Usage/UsageQuery.php b/src/Usage/UsageQuery.php
index 1976a13..cb4b507 100644
--- a/src/Usage/UsageQuery.php
+++ b/src/Usage/UsageQuery.php
@@ -26,11 +26,31 @@
  * switches from raw row returns to aggregated results grouped by time bucket:
  * - Events: SUM(value) per bucket
  * - Gauges: argMax(value, time) per bucket
+ *
+ * An `aggregate` hint overrides how values are combined: `max` takes the
+ * largest value per bucket, which is how a gauge level series rolls up to a
+ * coarser interval — the default `argMax` would return the latest reading
+ * rather than the highest.
  */
 class UsageQuery extends Query
 {
     public const TYPE_GROUP_BY_INTERVAL = 'groupByInterval';
     public const TYPE_GROUP_BY = 'groupBy';
+    public const TYPE_AGGREGATE = 'aggregate';
+
+    /**
+     * Valid aggregation functions.
+     *
+     * - `max` — the largest value per bucket, overriding the per-type default.
+     *   Intended for gauges, where the default `argMax(value, time)` returns the
+     *   *latest* reading rather than the highest one. Reading a pre-computed
+     *   level series (e.g. realtime concurrency sampled every 5 minutes) at a
+     *   coarser interval needs the peak of the samples, not the last one.
+     *
+     * There is deliberately no `sum`: it is already the default for events, and
+     * on gauges it would total point-in-time snapshots, which means nothing.
+     */
+    public const VALID_AGGREGATES = ['max'];
 
     /**
      * Valid interval values and their ClickHouse INTERVAL equivalents.
@@ -51,7 +71,7 @@ class UsageQuery extends Query
      */
     public static function isMethod(string $value): bool
     {
-        if ($value === self::TYPE_GROUP_BY_INTERVAL || $value === self::TYPE_GROUP_BY) {
+        if ($value === self::TYPE_GROUP_BY_INTERVAL || $value === self::TYPE_GROUP_BY || $value === self::TYPE_AGGREGATE) {
             return true;
         }
 
@@ -180,4 +200,72 @@ public static function removeGroupBy(array $queries): array
             return !self::isGroupBy($query);
         }));
     }
+
+    /**
+     * Create an aggregate query selecting the aggregation function.
+     *
+     * `max` takes the largest value in the bucket, overriding the per-type
+     * default — the meaningful roll-up for a gauge level series, whose default
+     * `argMax(value, time)` would return the latest reading.
+     * See {@see UsageQuery::VALID_AGGREGATES}.
+     *
+     * @param string $function One of {@see UsageQuery::VALID_AGGREGATES}.
+     * @return self
+     */
+    public static function aggregate(string $function): self
+    {
+        if (!in_array($function, self::VALID_AGGREGATES, true)) {
+            throw new \InvalidArgumentException(
+                "Invalid aggregate '{$function}'. Allowed: " . implode(', ', self::VALID_AGGREGATES)
+            );
+        }
+
+        return new self(self::TYPE_AGGREGATE, 'value', [$function]);
+    }
+
+    /**
+     * Check if a query is an aggregate query.
+     *
+     * @param Query $query
+     * @return bool
+     */
+    public static function isAggregate(Query $query): bool
+    {
+        return $query->getMethod() === self::TYPE_AGGREGATE;
+    }
+
+    /**
+     * Extract the aggregation function from an array of queries, if present.
+     *
+     * Queries parsed via `Query::parse()` are base `Query` objects rather than
+     * `UsageQuery` instances, so we match on the method string alone.
+     *
+     * @param array<Query> $queries
+     * @return string|null The aggregation function, or null if not present.
+     */
+    public static function extractAggregate(array $queries): ?string
+    {
+        foreach ($queries as $query) {
+            if (self::isAggregate($query)) {
+                $value = $query->getValues()[0] ?? null;
+
+                return is_string($value) ? $value : null;
+            }
+        }
+
+        return null;
+    }
+
+    /**
+     * Remove all aggregate queries from an array of queries.
+     *
+     * @param array<Query> $queries
+     * @return array<Query>
+     */
+    public static function removeAggregate(array $queries): array
+    {
+        return array_values(array_filter($queries, function (Query $query) {
+            return !self::isAggregate($query);
+        }));
+    }
 }
diff --git a/tests/Usage/AccumulatorTest.php b/tests/Usage/AccumulatorTest.php
index 8495c48..a2417df 100644
--- a/tests/Usage/AccumulatorTest.php
+++ b/tests/Usage/AccumulatorTest.php
@@ -275,13 +275,84 @@ public function testTagOrderDoesNotSplitEntries(): void
         $this->assertEquals(30, $this->adapter->batches[0]['metrics'][0]['value']);
     }
 
-    public function testNegativeValueThrows(): void
+    public function testGaugeNegativeValueThrows(): void
     {
+        // Gauges are snapshots (e.g. storage); callers never opt in, so
+        // negatives still throw by default.
+        $this->expectException(\InvalidArgumentException::class);
+        $this->expectExceptionMessage('Value cannot be negative');
+        $this->accumulator->collect('t1', 'storage', -1, Usage::TYPE_GAUGE);
+    }
+
+    public function testNegativeValueRejectedByDefaultForEvents(): void
+    {
+        // Default is strict for every metric, events included, so a buggy
+        // negative count is caught unless the caller explicitly opts in.
         $this->expectException(\InvalidArgumentException::class);
         $this->expectExceptionMessage('Value cannot be negative');
         $this->accumulator->collect('t1', 'requests', -1, Usage::TYPE_EVENT);
     }
 
+    public function testEventNegativeValueAllowedWhenOptedIn(): void
+    {
+        // Genuine signed deltas (realtime connections emit +1/-1) opt in per
+        // call with allowNegative, so a lone -1 persists rather than throws.
+        $this->accumulator->collect('t1', 'realtime.connections', -1, Usage::TYPE_EVENT, allowNegative: true);
+
+        $this->assertEquals(1, $this->accumulator->count());
+        $this->assertTrue($this->accumulator->flush());
+        $entry = $this->adapter->batches[0]['metrics'][0];
+        $this->assertEquals(-1, $entry['value']);
+        // The opt-in flag rides along on the row so addBatch can honour it.
+        $this->assertTrue($entry['allowNegative']);
+    }
+
+    public function testEventMixedSignsNetToDelta(): void
+    {
+        // +1,+1,-1 folds to a net delta of +1 for the metric.
+        $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, allowNegative: true);
+        $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT, allowNegative: true);
+        $this->accumulator->collect('t1', 'realtime.connections', -1, Usage::TYPE_EVENT, allowNegative: true);
+
+        $this->assertEquals(1, $this->accumulator->count());
+        $this->assertTrue($this->accumulator->flush());
+        $this->assertEquals(1, $this->adapter->batches[0]['metrics'][0]['value']);
+    }
+
+    public function testFoldedEntryKeepsTheNegativeOptIn(): void
+    {
+        // The opt-in belongs to the folded row, not to whichever call opened
+        // it. Here the entry is created by a plain positive with the flag off,
+        // then a signed delta folds in and takes the net negative.
+        $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT);
+        $this->accumulator->collect('t1', 'realtime.connections', -3, Usage::TYPE_EVENT, allowNegative: true);
+
+        $this->assertEquals(1, $this->accumulator->count());
+        $this->assertTrue($this->accumulator->flush());
+
+        $entry = $this->adapter->batches[0]['metrics'][0];
+        $this->assertEquals(-2, $entry['value']);
+
+        // Without this the net row is written unauthorised and rejected at
+        // validation. A failed batch keeps its entries buffered, so every
+        // later flush would retry the same rejection.
+        $this->assertTrue($entry['allowNegative']);
+    }
+
+    public function testFoldRetainsOptInRegardlessOfCallOrder(): void
+    {
+        // Same fold, opposite order: the entry is opened by the opted-in
+        // delta and a plain positive folds in afterwards.
+        $this->accumulator->collect('t1', 'realtime.connections', -3, Usage::TYPE_EVENT, allowNegative: true);
+        $this->accumulator->collect('t1', 'realtime.connections', 1, Usage::TYPE_EVENT);
+
+        $this->assertTrue($this->accumulator->flush());
+
+        $entry = $this->adapter->batches[0]['metrics'][0];
+        $this->assertEquals(-2, $entry['value']);
+        $this->assertTrue($entry['allowNegative']);
+    }
+
     public function testInvalidTypeThrows(): void
     {
         $this->expectException(\InvalidArgumentException::class);
diff --git a/tests/Usage/Adapter/ClickHouseGaugeMaxTest.php b/tests/Usage/Adapter/ClickHouseGaugeMaxTest.php
new file mode 100644
index 0000000..882ec4d
--- /dev/null
+++ b/tests/Usage/Adapter/ClickHouseGaugeMaxTest.php
@@ -0,0 +1,264 @@
+<?php
+
+namespace Utopia\Tests\Usage\Adapter;
+
+use Utopia\Query\Query;
+use Utopia\Usage\Adapter\ClickHouse as ClickHouseAdapter;
+use Utopia\Usage\Usage;
+use Utopia\Usage\UsageQuery;
+
+/**
+ * Integration tests for `aggregate('max')` on the gauge read path.
+ *
+ * Gauges default to `argMax(value, time)` — the *latest* reading in the bucket.
+ * That is right for a snapshot ("how much storage is used now"), but wrong for
+ * rolling a sampled level series up to a coarser interval: reading a realtime
+ * concurrency gauge sampled every 5 minutes at a 1h interval should give the
+ * hour's highest sample, not its last one.
+ *
+ * Rows are inserted with raw SQL so the query under test stays isolated from
+ * the write path.
+ */
+class ClickHouseGaugeMaxTest extends ClickHouseTestCase
+{
+    private Usage $usage;
+
+    private ClickHouseAdapter $adapter;
+
+    protected function setUp(): void
+    {
+        $host = getenv('CLICKHOUSE_HOST') ?: 'clickhouse';
+        $username = getenv('CLICKHOUSE_USER') ?: 'default';
+        $password = getenv('CLICKHOUSE_PASSWORD') ?: 'clickhouse';
+        $port = (int) (getenv('CLICKHOUSE_PORT') ?: 8123);
+        $secure = (bool) (getenv('CLICKHOUSE_SECURE') ?: false);
+
+        $this->adapter = new ClickHouseAdapter(
+            $host,
+            $username,
+            $password,
+            $port,
+            $secure,
+            namespace: 'utopia_usage_gauge_max',
+            database: getenv('CLICKHOUSE_DATABASE') ?: 'default',
+        );
+
+        $this->usage = new Usage($this->adapter);
+        $this->usage->setup();
+    }
+
+    /**
+     * Insert gauge samples straight into the gauges table.
+     *
+     * @param  array<int, array{value: int, time: string}>  $rows
+     */
+    private function insertSamples(string $metric, array $rows): void
+    {
+        $table = $this->resolveTableName($this->adapter, 'getGaugesTableName');
+        $database = $this->databaseName($this->adapter);
+        $ref = '`'.$database.'`.`'.$table.'`';
+
+        $tuples = [];
+        foreach ($rows as $i => $row) {
+            $id = $metric.'-'.$i;
+            $value = (int) $row['value'];
+            $time = $row['time'];
+            $tuples[] = "('{$id}', '{$metric}', {$value}, '{$time}')";
+        }
+
+        $sql = "INSERT INTO {$ref} (id, metric, value, time) VALUES ".implode(', ', $tuples);
+        $this->queryRaw($this->adapter, $sql);
+    }
+
+    public function test_gauge_defaults_to_latest_reading(): void
+    {
+        $metric = 'rt-concurrent-default-'.uniqid();
+
+        // Peak is 9, but the last sample is 4 — the default argMax path
+        // must still return the latest reading.
+        $this->insertSamples($metric, [
+            ['value' => 2, 'time' => '2026-06-01 00:00:00'],
+            ['value' => 9, 'time' => '2026-06-01 00:05:00'],
+            ['value' => 4, 'time' => '2026-06-01 00:10:00'],
+        ]);
+
+        $results = $this->usage->find('1', [
+            Query::equal('metric', [$metric]),
+            Query::greaterThanEqual('time', '2026-06-01 00:00:00'),
+            Query::lessThanEqual('time', '2026-06-01 01:00:00'),
+            UsageQuery::groupByInterval('time', '1h'),
+        ], Usage::TYPE_GAUGE);
+
+        $this->assertCount(1, $results);
+        $this->assertEquals(4, $results[0]->getValue());
+    }
+
+    public function test_gauge_max_returns_highest_sample(): void
+    {
+        $metric = 'rt-concurrent-max-'.uniqid();
+
+        $this->insertSamples($metric, [
+            ['value' => 2, 'time' => '2026-06-01 00:00:00'],
+            ['value' => 9, 'time' => '2026-06-01 00:05:00'],
+            ['value' => 4, 'time' => '2026-06-01 00:10:00'],
+        ]);
+
+        $results = $this->usage->find('1', [
+            Query::equal('metric', [$metric]),
+            Query::greaterThanEqual('time', '2026-06-01 00:00:00'),
+            Query::lessThanEqual('time', '2026-06-01 01:00:00'),
+            UsageQuery::groupByInterval('time', '1h'),
+            UsageQuery::aggregate('max'),
+        ], Usage::TYPE_GAUGE);
+
+        $this->assertCount(1, $results);
+        $this->assertEquals(9, $results[0]->getValue());
+    }
+
+    public function test_gauge_max_per_bucket(): void
+    {
+        $metric = 'rt-concurrent-buckets-'.uniqid();
+
+        // Hour 00 peaks at 9; hour 01 peaks at 6. Max composes upward, so
+        // each bucket reports its own highest sample.
+        $this->insertSamples($metric, [
+            ['value' => 2, 'time' => '2026-06-01 00:00:00'],
+            ['value' => 9, 'time' => '2026-06-01 00:05:00'],
+            ['value' => 4, 'time' => '2026-06-01 00:10:00'],
+            ['value' => 6, 'time' => '2026-06-01 01:05:00'],
+            ['value' => 1, 'time' => '2026-06-01 01:10:00'],
+        ]);
+
+        $results = $this->usage->find('1', [
+            Query::equal('metric', [$metric]),
+            Query::greaterThanEqual('time', '2026-06-01 00:00:00'),
+            Query::lessThanEqual('time', '2026-06-01 02:00:00'),
+            UsageQuery::groupByInterval('time', '1h'),
+            UsageQuery::aggregate('max'),
+        ], Usage::TYPE_GAUGE);
+
+        $this->assertCount(2, $results);
+        $this->assertEquals(9, $results[0]->getValue());
+        $this->assertEquals(6, $results[1]->getValue());
+    }
+
+    public function test_gauge_max_flat_aggregate_over_window(): void
+    {
+        $metric = 'rt-concurrent-flat-'.uniqid();
+
+        // No interval: one row for the whole window — the billing shape.
+        $this->insertSamples($metric, [
+            ['value' => 2, 'time' => '2026-06-01 00:00:00'],
+            ['value' => 9, 'time' => '2026-06-01 00:05:00'],
+            ['value' => 6, 'time' => '2026-06-01 01:05:00'],
+        ]);
+
+        $results = $this->usage->find('1', [
+            Query::equal('metric', [$metric]),
+            Query::greaterThanEqual('time', '2026-06-01 00:00:00'),
+            Query::lessThanEqual('time', '2026-06-01 02:00:00'),
+            UsageQuery::aggregate('max'),
+        ], Usage::TYPE_GAUGE);
+
+        $this->assertCount(1, $results);
+        $this->assertEquals(9, $results[0]->getValue());
+    }
+
+    /**
+     * Shared-tables adapter — the multi-tenant shape cloud runs. Built on
+     * demand so the single-tenant tests above keep the simpler schema.
+     *
+     * @return array{0: Usage, 1: ClickHouseAdapter}
+     */
+    private function sharedTablesUsage(): array
+    {
+        $adapter = new ClickHouseAdapter(
+            getenv('CLICKHOUSE_HOST') ?: 'clickhouse',
+            getenv('CLICKHOUSE_USER') ?: 'default',
+            getenv('CLICKHOUSE_PASSWORD') ?: 'clickhouse',
+            (int) (getenv('CLICKHOUSE_PORT') ?: 8123),
+            (bool) (getenv('CLICKHOUSE_SECURE') ?: false),
+            namespace: 'utopia_usage_gauge_max_shared',
+            database: getenv('CLICKHOUSE_DATABASE') ?: 'default',
+            sharedTables: true,
+        );
+
+        $usage = new Usage($adapter);
+        $usage->setup();
+
+        return [$usage, $adapter];
+    }
+
+    /**
+     * Two tenants, one table, one metric — 7 for tenant-a and 5 for tenant-b.
+     */
+    private function seedTwoTenants(ClickHouseAdapter $adapter, string $metric): void
+    {
+        $table = $this->resolveTableName($adapter, 'getGaugesTableName');
+        $database = $this->databaseName($adapter);
+        $ref = '`'.$database.'`.`'.$table.'`';
+
+        $sql = "INSERT INTO {$ref} (id, metric, value, time, tenant) VALUES "
+            ."('{$metric}-a1', '{$metric}', 3, '2026-06-01 00:00:00', 'tenant-a'), "
+            ."('{$metric}-a2', '{$metric}', 7, '2026-06-01 00:05:00', 'tenant-a'), "
+            ."('{$metric}-b1', '{$metric}', 2, '2026-06-01 00:00:00', 'tenant-b'), "
+            ."('{$metric}-b2', '{$metric}', 5, '2026-06-01 00:05:00', 'tenant-b')";
+        $this->queryRaw($adapter, $sql);
+    }
+
+    public function test_gauge_max_stays_scoped_to_one_tenant(): void
+    {
+        [$usage, $adapter] = $this->sharedTablesUsage();
+        $metric = 'rt-concurrent-scoped-'.uniqid();
+        $this->seedTwoTenants($adapter, $metric);
+
+        // tenant-a's own peak — never tenant-b's, never a combined figure.
+        $results = $usage->find('tenant-a', [
+            Query::equal('metric', [$metric]),
+            Query::greaterThanEqual('time', '2026-06-01 00:00:00'),
+            Query::lessThanEqual('time', '2026-06-01 01:00:00'),
+            UsageQuery::aggregate('max'),
+        ], Usage::TYPE_GAUGE);
+
+        $this->assertCount(1, $results);
+        $this->assertEquals(7, $results[0]->getValue());
+    }
+
+    public function test_find_across_tenants_returns_a_row_per_tenant(): void
+    {
+        [$usage, $adapter] = $this->sharedTablesUsage();
+        $metric = 'rt-concurrent-crosstenant-'.uniqid();
+        $this->seedTwoTenants($adapter, $metric);
+
+        // One pass over every tenant — what the aggregation job needs instead
+        // of N per-tenant queries. groupBy('tenant') keeps rows attributable.
+        $results = $usage->findAcrossTenants([
+            Query::equal('metric', [$metric]),
+            Query::greaterThanEqual('time', '2026-06-01 00:00:00'),
+            Query::lessThanEqual('time', '2026-06-01 01:00:00'),
+            UsageQuery::groupBy('tenant'),
+            UsageQuery::aggregate('max'),
+        ], Usage::TYPE_GAUGE);
+
+        $byTenant = [];
+        foreach ($results as $row) {
+            $tenant = $row->getTenant();
+            $this->assertNotNull($tenant, 'every cross-tenant row must carry its tenant');
+            $byTenant[$tenant] = $row->getValue();
+        }
+
+        $this->assertCount(2, $byTenant);
+        $this->assertEquals(7, $byTenant['tenant-a']);
+        $this->assertEquals(5, $byTenant['tenant-b']);
+    }
+
+    public function test_find_across_tenants_rejected_without_shared_tables(): void
+    {
+        $this->expectException(\Exception::class);
+        $this->expectExceptionMessage('requires shared-tables mode');
+
+        $this->usage->findAcrossTenants([
+            Query::equal('metric', ['anything']),
+        ], Usage::TYPE_GAUGE);
+    }
+}
diff --git a/tests/Usage/Adapter/DatabaseTest.php b/tests/Usage/Adapter/DatabaseTest.php
index 3352b3a..4dfcb80 100644
--- a/tests/Usage/Adapter/DatabaseTest.php
+++ b/tests/Usage/Adapter/DatabaseTest.php
@@ -9,6 +9,7 @@
 use Utopia\Database\Adapter\MariaDB;
 use Utopia\Database\Database;
 use Utopia\Database\Exception\Duplicate;
+use Utopia\Query\Query;
 use Utopia\Tests\Usage\UsageBase;
 use Utopia\Usage\Adapter\Database as AdapterDatabase;
 use Utopia\Usage\Usage;
@@ -259,4 +260,40 @@ public function testHealthCheckWithNonExistentDatabase(): void
             $this->assertNotEmpty($health['error']);
         }
     }
+
+    public function testNegativeValueRejectedByDefault(): void
+    {
+        if (!extension_loaded('pdo_mysql')) {
+            $this->markTestSkipped('pdo_mysql extension is not installed');
+        }
+
+        $this->expectException(\InvalidArgumentException::class);
+        $this->expectExceptionMessage('Value cannot be negative');
+
+        $this->usage->addBatch([
+            ['tenant' => '1', 'metric' => 'db-negative-default', 'value' => -1],
+        ], Usage::TYPE_EVENT);
+    }
+
+    public function testNegativeValuePersistsWhenOptedIn(): void
+    {
+        if (!extension_loaded('pdo_mysql')) {
+            $this->markTestSkipped('pdo_mysql extension is not installed');
+        }
+
+        $this->usage->purge('1', [], Usage::TYPE_EVENT);
+
+        // A signed delta opts in per row. The accumulator sets this flag and
+        // the ClickHouse adapter honours it; an adapter that ignored it would
+        // reject the row on every flush, since a failed batch stays buffered.
+        $this->assertTrue($this->usage->addBatch([
+            ['tenant' => '1', 'metric' => 'db-negative-optin', 'value' => -3, 'allowNegative' => true],
+        ], Usage::TYPE_EVENT));
+
+        $this->assertEquals(
+            -3,
+            $this->usage->sum('1', [Query::equal('metric', ['db-negative-optin'])], 'value', Usage::TYPE_EVENT),
+            'the opted-in negative must be stored, not silently dropped',
+        );
+    }
 }
diff --git a/tests/Usage/UsageQueryTest.php b/tests/Usage/UsageQueryTest.php
index 924c6e3..3b41106 100644
--- a/tests/Usage/UsageQueryTest.php
+++ b/tests/Usage/UsageQueryTest.php
@@ -231,4 +231,96 @@ public function testExtractGroupByFromParsedQuery(): void
         $this->assertCount(1, $extracted);
         $this->assertEquals('service', $extracted[0]->getAttribute());
     }
+
+    public function testAggregateCreation(): void
+    {
+        $query = UsageQuery::aggregate('max');
+
+        $this->assertInstanceOf(UsageQuery::class, $query);
+        $this->assertEquals(UsageQuery::TYPE_AGGREGATE, $query->getMethod());
+        $this->assertEquals(['max'], $query->getValues());
+        $this->assertEquals('max', $query->getValue());
+    }
+
+    public function testAggregateAcceptsAllValidFunctions(): void
+    {
+        foreach (UsageQuery::VALID_AGGREGATES as $function) {
+            $query = UsageQuery::aggregate($function);
+            $this->assertEquals($function, $query->getValue());
+        }
+    }
+
+    public function testAggregateRejectsInvalidFunction(): void
+    {
+        $this->expectException(\InvalidArgumentException::class);
+        $this->expectExceptionMessage("Invalid aggregate 'peak'");
+        UsageQuery::aggregate('peak');
+    }
+
+    public function testAggregateIsMethod(): void
+    {
+        $this->assertTrue(UsageQuery::isMethod(UsageQuery::TYPE_AGGREGATE));
+    }
+
+    public function testIsAggregate(): void
+    {
+        $aggregate = UsageQuery::aggregate('max');
+        $regular = Query::equal('metric', ['bandwidth']);
+
+        $this->assertTrue(UsageQuery::isAggregate($aggregate));
+        $this->assertFalse(UsageQuery::isAggregate($regular));
+    }
+
+    public function testExtractAggregate(): void
+    {
+        $queries = [
+            Query::equal('metric', ['realtime.connections']),
+            UsageQuery::aggregate('max'),
+        ];
+
+        $this->assertEquals('max', UsageQuery::extractAggregate($queries));
+    }
+
+    public function testExtractAggregateFromParsedQuery(): void
+    {
+        // Queries created via Query::parse() are base Query objects, not UsageQuery.
+        $parsedAggregate = new Query(UsageQuery::TYPE_AGGREGATE, 'value', ['max']);
+        $equal = Query::equal('metric', ['realtime.connections']);
+
+        $this->assertEquals('max', UsageQuery::extractAggregate([$equal, $parsedAggregate]));
+    }
+
+    public function testExtractAggregateReturnsNullWhenMissing(): void
+    {
+        $queries = [
+            Query::equal('metric', ['realtime.connections']),
+            UsageQuery::groupByInterval('time', '1h'),
+        ];
+
+        $this->assertNull(UsageQuery::extractAggregate($queries));
+    }
+
+    public function testRemoveAggregate(): void
+    {
+        $queries = [
+            Query::equal('metric', ['realtime.connections']),
+            UsageQuery::aggregate('max'),
+            UsageQuery::groupByInterval('time', '1h'),
+        ];
+
+        $remaining = UsageQuery::removeAggregate($queries);
+
+        $this->assertCount(2, $remaining);
+        foreach ($remaining as $query) {
+            $this->assertNotEquals(UsageQuery::TYPE_AGGREGATE, $query->getMethod());
+        }
+    }
+
+    public function testValidAggregatesConstant(): void
+    {
+        // `max` is the only selectable aggregate: it overrides the per-type
+        // default. `sum` is absent on purpose - already the default for events,
+        // and on gauges it would total point-in-time snapshots.
+        $this->assertSame(['max'], UsageQuery::VALID_AGGREGATES);
+    }
 }