feat(telemetry): core-node observability foundation — OTLP logs + traces + metrics

.gitignore

@@ -45,6 +45,7 @@ packages/evm-module/typechain/
 
 packages/evm-module/deployments/hardhat_contracts.json
 packages/evm-module/deployments/localhost_contracts.json
+packages/evm-module/deployments/localhost/
 snapshots/_cache_phase1_neuroweb_epoch16.json
 .claude/
 .orchestrator/

packages/agent/package.json

@@ -14,6 +14,7 @@
     "clean": "rm -rf dist tsconfig.tsbuildinfo"
   },
   "dependencies": {
+    "@opentelemetry/api": "^1.9.1",
     "@libp2p/peer-id": "^6.0.9",
     "@multiformats/multiaddr": "^13.0.3",
     "@noble/ciphers": "^2.2.0",

packages/agent/src/dkg-agent-publish.ts

@@ -97,8 +97,11 @@ import {
   pickNetworkTunables,
   sharedMemoryReadBothFilter,
   partitionCatalogQuads,
+  withSpan,
+  getMetrics,
   assertQuadLiteralsMutf8Safe,
 } from '@origintrail-official/dkg-core';
+import { SpanStatusCode } from '@opentelemetry/api';
 import { GraphManager, PrivateContentStore, createTripleStore, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig } from '@origintrail-official/dkg-storage';
 import { EVMChainAdapter, NoChainAdapter, enrichEvmError, buildKnowledgeAssetUal, type EVMAdapterConfig, type ChainAdapter, type CreateContextGraphParams, type CreateOnChainContextGraphParams, type CreateOnChainContextGraphResult, type TxResult, type V10PublishingConvictionAccountInfo } from '@origintrail-official/dkg-chain';
 import {
@@ -1290,6 +1293,15 @@ export class PublishMethods extends DKGAgentBase {
     privateQuads?: Quad[],
     opts?: PublishOpts,
   ): Promise<PublishResult> {
+   return withSpan('agent.publish', async (span) => {
+    const chainId = typeof this.chain?.chainId === 'string' && this.chain.chainId !== 'none' ? this.chain.chainId : undefined;
+    const publishStartedAt = Date.now();
+    span.setAttributes({
+      'dkg.context_graph_id': contextGraphId,
+      'dkg.triple_count': quads.length,
+      'dkg.has_private': !!privateQuads && privateQuads.length > 0,
+      ...(chainId ? { 'dkg.chain_id': chainId } : {}),
+    });
     const ctx = opts?.operationCtx ?? createOperationContext('publish');
     const onPhase = opts?.onPhase;
     this.log.info(ctx, `Starting publish to context graph "${contextGraphId}" with ${quads.length} triples`);
@@ -1425,6 +1437,12 @@ export class PublishMethods extends DKGAgentBase {
       encryptInlineChunked,
     });
 
+    span.setAttribute('dkg.publish_status', result.status);
+    if (result.status === 'failed') {
+      span.setStatus({ code: SpanStatusCode.ERROR });
+      span.addEvent('publish_failed', { error: String(result.contextGraphError ?? '') });
+    }
+
     onPhase?.('broadcast', 'start');
     this.log.info(ctx, `Local publish complete, broadcasting to peers`);
     await this.broadcastPublish(contextGraphId, result, ctx);
@@ -1436,7 +1454,12 @@ export class PublishMethods extends DKGAgentBase {
     // it can never affect the publish just completed.
     await this.emitPublicProjectionAfterPublish(contextGraphId, result, ctx);
 
+    const publishMetricAttrs = { outcome: result.status, source: 'direct', ...(chainId ? { chain_id: chainId } : {}) };
+    getMetrics().publishTotal.add(1, publishMetricAttrs);
+    getMetrics().publishDuration.record(Date.now() - publishStartedAt, publishMetricAttrs);
+
     return result;
+   });
   }
 
   /**
@@ -4117,6 +4140,14 @@ export class PublishMethods extends DKGAgentBase {
       schemeVersion?: number;
     },
   ): Promise<PublishResult> {
+   return withSpan('agent.publish_from_swm', async (span) => {
+    const chainId = typeof this.chain?.chainId === 'string' && this.chain.chainId !== 'none' ? this.chain.chainId : undefined;
+    const publishStartedAt = Date.now();
+    span.setAttributes({
+      'dkg.context_graph_id': contextGraphId,
+      'dkg.selection': selection === 'all' ? 'all' : 'roots',
+      ...(chainId ? { 'dkg.chain_id': chainId } : {}),
+    });
     const ctx = options?.operationCtx ?? createOperationContext('publishFromSWM');
     const effectiveSubCG = options?.subContextGraphId ?? options?.contextGraphId;
     // `ctxGraphIdStr` doubles as `publishContextGraphId` for REMAP-flow
@@ -4248,6 +4279,12 @@ export class PublishMethods extends DKGAgentBase {
       encryptInlineChunked,
     });
 
+    span.setAttribute('dkg.publish_status', result.status);
+    if (result.status === 'failed') {
+      span.setStatus({ code: SpanStatusCode.ERROR });
+      span.addEvent('publish_failed', { error: String(result.contextGraphError ?? '') });
+    }
+
     if (result.status === 'confirmed' && result.onChainResult) {
       const rootEntities = result.kaManifest.map(ka => ka.rootEntity);
 
@@ -4338,7 +4375,12 @@ export class PublishMethods extends DKGAgentBase {
       }
     }
 
+    const publishMetricAttrs = { outcome: result.status, source: 'swm', ...(chainId ? { chain_id: chainId } : {}) };
+    getMetrics().publishTotal.add(1, publishMetricAttrs);
+    getMetrics().publishDuration.record(Date.now() - publishStartedAt, publishMetricAttrs);
+
     return result;
+   });
   }
 
   /** @deprecated Use publishFromSharedMemory. Will be removed in V10.1. */

packages/agent/src/p2p/sync-transport.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from 'node:crypto';
-import { withRetry } from '@origintrail-official/dkg-core';
+import { withRetry, withSpan, getMetrics } from '@origintrail-official/dkg-core';
 import { markSyncTransportFailure } from '../sync/error-tags.js';
 
 /**
@@ -82,7 +82,11 @@ interface SyncSendParams {
 }
 
 export async function sendSyncRequest(params: SyncSendParams): Promise<Uint8Array> {
-  return withRetry(
+  return withSpan(
+    'sync.request',
+    async () => {
+      try {
+        const out = await withRetry(
     async () => {
       throwIfAborted(params.signal);
       const requestBytes = await params.requestFactory();
@@ -114,6 +118,15 @@ export async function sendSyncRequest(params: SyncSendParams): Promise<Uint8Arra
       isRetryable: () => params.signal?.aborted !== true,
       onRetry: params.onRetry,
     },
+        );
+        getMetrics().syncRequestTotal.add(1, { outcome: 'ok', protocol_id: params.protocolId });
+        return out;
+      } catch (err) {
+        getMetrics().syncRequestTotal.add(1, { outcome: 'error', protocol_id: params.protocolId });
+        throw err;
+      }
+    },
+    { attributes: { 'dkg.protocol_id': params.protocolId } },
   );
 }
 

packages/agent/src/sync/responder/sync-handler.ts

@@ -1,6 +1,8 @@
 import {
   createOperationContext,
   QuietRetryableHandlerError,
+  withSpan,
+  getMetrics,
   type OperationContext,
 } from '@origintrail-official/dkg-core';
 import type { TripleStore } from '@origintrail-official/dkg-storage';
@@ -297,7 +299,8 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void {
     };
   };
 
-  register(protocolSync, async (data, peerId, options) => {
+  register(protocolSync, async (data, peerId, options) => withSpan('sync.response', async (span) => {
+    span.setAttribute('dkg.protocol_id', protocolSync);
     const signal = options?.signal;
     const handlerStartedAt = Date.now();
     const request = parseSyncRequest(data);
@@ -495,12 +498,19 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void {
         logDebug(createOperationContext('sync'), `Sync responder total for "${contextGraphId}" (phase=${phase}, workspace=${isWorkspace}): ${totalDurationMs}ms`);
       }
       return new TextEncoder().encode(nquads.join('\n'));
+    }).then((res) => {
+      getMetrics().syncResponseTotal.add(1, { outcome: 'ok' });
+      return res;
     }).catch((err) => {
       if (err instanceof SyncResponderBusyError) {
+        getMetrics().syncResponseTotal.add(1, { outcome: 'busy' });
+        span.setAttribute('dkg.sync_response_outcome', 'busy');
         logDebug(createOperationContext('sync'), `Sync responder busy for "${contextGraphId}" from peer ${peerId} (phase=${phase}): ${err.message}`);
         throw new QuietRetryableHandlerError(err.message);
       }
       if (err instanceof SyncRowSnapshotLimitError) {
+        getMetrics().syncResponseTotal.add(1, { outcome: 'limit' });
+        span.setAttribute('dkg.sync_response_outcome', 'limit');
         logWarn(
           createOperationContext('sync'),
           `Sync responder snapshot limit for "${contextGraphId}" from peer ${peerId} (phase=${phase}, workspace=${isWorkspace}): active=${err.activeEntries}/${err.maxEntries} cached=${err.cachedEntries} inflight=${err.inflightEntries} key=${err.key}`,
@@ -509,7 +519,8 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void {
           `sync responder snapshot limit exceeded (active=${err.activeEntries}/${err.maxEntries})`,
         );
       }
+      getMetrics().syncResponseTotal.add(1, { outcome: 'error' });
       throw err;
     });
-  });
+  }));
 }