Chapter 5: SDK Architecture — Universal Agent Client Libraries for Any Runtime

ALGORITHM: InitializeAgentClient(config: SdkConfig) → AgentClient
 
1.  VALIDATE config against SdkConfigSchema
    IF validation fails → RAISE InvalidRequest("Configuration schema violation", details)
 
2.  RESOLVE transport := SelectTransport(config.endpoint, config.transport_preference)
    // Returns: GRPC | JSON_RPC | MCP_STDIO | MCP_SSE | WEBSOCKET | IN_PROCESS
 
3.  CONSTRUCT wire_layer := WireLayer.create(transport, config.tls, config.compression)
4.  CONSTRUCT protocol_layer := ProtocolLayer.create(transport, wire_layer)
 
5.  CONSTRUCT connection_pool := ConnectionPool.create(
        min_connections = config.pool.min_size,
        max_connections = config.pool.max_size,
        idle_timeout    = config.pool.idle_timeout,
        health_check_interval = config.pool.health_interval
    )
 
6.  CONSTRUCT middleware_chain := MiddlewareChain.build([
        AuthInterceptor(config.auth),
        RetryInterceptor(config.retry_policy),
        RateLimitInterceptor(config.rate_limit),
        CircuitBreakerInterceptor(config.circuit_breaker),
        LoggingInterceptor(config.logging),
        MetricsInterceptor(config.metrics),
        TracingInterceptor(config.tracing),
        DeadlineInterceptor(config.default_deadline)
    ])
 
7.  CONSTRUCT client := AgentClient(
        protocol    = protocol_layer,
        pool        = connection_pool,
        middleware   = middleware_chain,
        tool_registry = ToolRegistry(protocol_layer, config.tool_discovery),
        memory_store  = MemoryStore(protocol_layer, config.memory),
        context_builder = ContextBuilder(config.context),
        orchestrator = OrchestratorHandle(protocol_layer, config.orchestrator)
    )
 
8.  PERFORM client.health_check() with timeout = config.init_timeout
    IF health_check fails AND config.fail_fast = true → RAISE Unavailable(...)
    IF health_check fails AND config.fail_fast = false → LOG warning, SET client.degraded = true
 
9.  EMIT telemetry event: SDK_INITIALIZED {
        transport, pool_size, middleware_count, degraded_state, latency_ms
    }
 
10. RETURN client

OrchestratorHandle
    ├── AgentClient
    │     ├── ToolRegistry
    │     ├── MemoryStore
    │     └── ContextBuilder
    │           ├── ToolRegistry (schema retrieval)
    │           └── MemoryStore  (memory summaries)
    └── ContextBuilder (plan context assembly)

TYPE AgentClient:
    // Lifecycle
    FUNC connect(config: ConnectionConfig) → Result<void, SdkError>
    FUNC disconnect(grace_period: Duration) → Result<void, SdkError>
    FUNC health() → Result<HealthStatus, SdkError>
 
    // Core Agent Operations
    FUNC execute(request: AgentRequest, deadline: Duration) → Result<AgentResponse, SdkError>
    FUNC stream_execute(request: AgentRequest, deadline: Duration) → AsyncStream<AgentChunk, SdkError>
 
    // Plan-Act-Verify Loop (Orchestrated)
    FUNC run_loop(task: TaskSpec, options: LoopOptions) → Result<TaskResult, SdkError>
 
    // Tool Operations (delegated)
    PROPERTY tools: ToolRegistry
    // Memory Operations (delegated)
    PROPERTY memory: MemoryStore
    // Context Operations (delegated)
    PROPERTY context: ContextBuilder
    // Orchestration (delegated)
    PROPERTY orchestrator: OrchestratorHandle
 
    // Middleware
    FUNC use(interceptor: Interceptor) → void

TYPE ToolRegistry:
    FUNC discover(filter: ToolFilter, pagination: PageRequest) → Result<Page<ToolDescriptor>, SdkError>
    FUNC get_schema(tool_id: ToolId, version: SemanticVersion?) → Result<ToolSchema, SdkError>
    FUNC invoke(tool_id: ToolId, input: TypedInput, options: InvokeOptions) → Result<ToolOutput, SdkError>
    FUNC invoke_streaming(tool_id: ToolId, input: TypedInput) → AsyncStream<ToolChunk, SdkError>
    FUNC subscribe_changes(filter: ToolFilter) → AsyncStream<ToolChangeEvent, SdkError>
    FUNC validate_input(tool_id: ToolId, input: JsonValue) → Result<ValidationResult, SdkError>
    FUNC cached_schemas() → Map<ToolId, CachedToolSchema>
    FUNC invalidate_cache(tool_id: ToolId?) → void

TYPE MemoryStore:
    // Read Operations
    FUNC query(query: MemoryQuery, tier: MemoryTier?) → Result<List<MemoryItem>, SdkError>
    FUNC get(key: MemoryKey, tier: MemoryTier) → Result<MemoryItem?, SdkError>
    FUNC summarize(scope: MemoryScope, budget: TokenBudget) → Result<MemorySummary, SdkError>
 
    // Write Operations (validated, provenance-tagged)
    FUNC write(item: MemoryWriteRequest) → Result<MemoryReceipt, SdkError>
    FUNC update(key: MemoryKey, patch: MemoryPatch, precondition: ETag?) → Result<MemoryReceipt, SdkError>
    FUNC delete(key: MemoryKey, reason: DeletionReason) → Result<void, SdkError>
 
    // Policy
    FUNC set_expiry(key: MemoryKey, policy: ExpiryPolicy) → Result<void, SdkError>
    FUNC deduplicate(scope: MemoryScope) → Result<DeduplicationReport, SdkError>

ALGORITHM: MemoryWriteAdmission(item: MemoryWriteRequest) → Result<MemoryReceipt, SdkError>
 
1.  VALIDATE item.schema against tier-specific MemoryItemSchema
    IF invalid → RETURN Err(InvalidRequest("Schema violation"))
 
2.  CHECK provenance:
    IF item.provenance IS EMPTY → RETURN Err(InvalidRequest("Provenance required"))
    VALIDATE item.provenance.source ∈ ALLOWED_SOURCES
 
3.  DEDUPLICATION check:
    existing := memory_index.find_similar(item.content, threshold = 0.95)
    IF existing IS NOT EMPTY:
        IF item.merge_policy = SKIP → RETURN Ok(existing.receipt)
        IF item.merge_policy = MERGE → item := merge(existing, item)
        IF item.merge_policy = REPLACE → mark existing as superseded
 
4.  EVALUATE expiry_policy:
    IF item.expiry IS UNDEFINED → item.expiry := tier_default_expiry(item.tier)
 
5.  COMPUTE item.embedding := embed(item.content) for semantic indexing
 
6.  PERSIST item to durable store with CAS (compare-and-swap) semantics
    IF CAS conflict → RETURN Err(ResourceExhausted("Concurrent write conflict"))
 
7.  EMIT event: MEMORY_WRITTEN { key, tier, provenance, size_tokens, expiry }
 
8.  RETURN Ok(MemoryReceipt { key, version, timestamp, tier })

TYPE ContextBuilder:
    FUNC set_role(policy: RolePolicy) → ContextBuilder
    FUNC set_task(objective: TaskObjective) → ContextBuilder
    FUNC add_retrieved(evidence: List<RetrievedChunk>) → ContextBuilder
    FUNC add_tool_affordances(tools: List<ToolDescriptor>) → ContextBuilder
    FUNC add_memory_summary(summary: MemorySummary) → ContextBuilder
    FUNC add_history(messages: List<Message>, compression: CompressionPolicy?) → ContextBuilder
    FUNC set_budget(budget: TokenBudget) → ContextBuilder
    FUNC compile() → Result<CompiledContext, SdkError>

ALGORITHM: AllocateTokenBudget(B: int, sections: List<Section>) → Map<SectionId, int>
 
1.  // Phase 1: Satisfy minimum allocations
    total_min := SUM(s.min_tokens FOR s IN sections)
    IF total_min > B → RAISE ResourceExhausted("Budget cannot satisfy minimums")
    remaining := B - total_min
 
2.  // Phase 2: Proportional allocation of remaining budget
    eligible := [s FOR s IN sections WHERE s.content_tokens > s.min_tokens]
    total_weight := SUM(s.priority_weight FOR s IN eligible)
 
    FOR EACH s IN eligible:
        s.bonus := FLOOR(remaining * (s.priority_weight / total_weight))
        s.allocated := MIN(s.min_tokens + s.bonus, s.content_tokens)
 
3.  // Phase 3: Redistribute unused tokens (sections at capacity)
    unused := remaining - SUM(s.bonus FOR s IN eligible)
    WHILE unused > 0 AND EXISTS s IN eligible WHERE s.allocated < s.content_tokens:
        distribute unused proportionally among under-capacity sections
        recalculate unused
 
4.  // Phase 4: Construct allocation map
    RETURN { s.id: s.allocated FOR s IN sections }

TYPE OrchestratorHandle:
    FUNC submit_task(task: TaskSpec) → Result<TaskHandle, SdkError>
    FUNC get_status(task_id: TaskId) → Result<TaskStatus, SdkError>
    FUNC cancel(task_id: TaskId, reason: CancellationReason) → Result<void, SdkError>
    FUNC subscribe(task_id: TaskId) → AsyncStream<TaskEvent, SdkError>
    FUNC list_agents(filter: AgentFilter) → Result<List<AgentDescriptor>, SdkError>
    FUNC claim_work_unit(agent_id: AgentId, unit_id: WorkUnitId, lease: Duration) → Result<Lease, SdkError>
    FUNC commit_result(lease: Lease, result: WorkUnitResult) → Result<CommitReceipt, SdkError>
    FUNC rollback(lease: Lease, reason: RollbackReason) → Result<void, SdkError>

ALGORITHM: AgentLoop(client: AgentClient, task: TaskSpec, opts: LoopOptions) → TaskResult
 
1.  plan := client.execute(PlanRequest(task, client.context.compile()), opts.deadline)
2.  steps := decompose(plan, max_steps = opts.max_steps)
 
3.  iteration := 0
4.  WHILE steps IS NOT EMPTY AND iteration < opts.max_iterations:
        step := steps.dequeue()
 
        // Retrieve
        evidence := client.context.retrieve(step.query, budget = opts.retrieval_budget)
 
        // Act
        action_result := MATCH step.type:
            TOOL_CALL → client.tools.invoke(step.tool_id, step.input, opts.tool_options)
            LLM_CALL  → client.execute(step.prompt, opts.llm_deadline)
            MEMORY_OP → client.memory.write(step.memory_item)
 
        // Verify
        verification := verify(action_result, step.expected_postconditions)
        IF verification.passed:
            // Critique (optional quality gate)
            critique := client.execute(CritiqueRequest(step, action_result), opts.critique_deadline)
            IF critique.severity < opts.critique_threshold:
                commit(step, action_result)
            ELSE:
                // Repair
                repair_steps := client.execute(RepairRequest(step, critique), opts.repair_deadline)
                steps.prepend(repair_steps)
        ELSE:
            // Rollback and retry or escalate
            IF step.retry_count < opts.max_step_retries:
                step.retry_count += 1
                steps.prepend(step)
            ELSE:
                persist_failure(step, verification)
                IF opts.on_failure = ABORT → RETURN TaskResult.failed(step, verification)
                IF opts.on_failure = SKIP  → CONTINUE
                IF opts.on_failure = ESCALATE → AWAIT human_approval(step)
 
        iteration += 1
 
5.  EMIT telemetry: AGENT_LOOP_COMPLETE { iterations, steps_completed, failures, total_tokens }
6.  RETURN TaskResult.success(committed_results)

IDLE ──acquire──→ ACTIVE ──release──→ IDLE
  │                  │
  │                  └──error──→ FAILED ──evict──→ CLOSED
  │
  └──health_fail──→ FAILED ──evict──→ CLOSED
  │
  └──shutdown──→ DRAINING ──drained──→ CLOSED

ALGORITHM: AcquireConnection(pool: ConnectionPool, deadline: Duration) → Result<Connection, SdkError>
 
1.  start_time := now()
 
2.  WHILE now() - start_time < deadline:
        // Try to acquire an idle connection
        c := pool.idle_queue.poll(timeout = MIN(100ms, deadline - elapsed))
 
        IF c IS NOT NULL:
            IF c.last_health_check > now() - pool.health_check_interval:
                c.state := ACTIVE
                RETURN Ok(c)
            ELSE:
                // Inline health check
                health := c.ping(timeout = pool.health_check_timeout)
                IF health.ok:
                    c.last_health_check := now()
                    c.state := ACTIVE
                    RETURN Ok(c)
                ELSE:
                    c.state := FAILED
                    pool.evict(c)
                    pool.metrics.increment("connection.health_fail")
                    CONTINUE
 
        // No idle connection available; try to create new
        IF pool.size() < pool.max_size:
            c := pool.create_new(timeout = deadline - elapsed)
            IF c IS NOT NULL:
                c.state := ACTIVE
                RETURN Ok(c)
 
        // Pool exhausted; apply backpressure
        pool.metrics.increment("connection.pool_exhausted")
 
3.  RETURN Err(DeadlineExceeded("Connection acquisition timed out"))

ALGORITHM: GracefulShutdown(client: AgentClient, grace_period: Duration)
 
1.  client.state := SHUTTING_DOWN
2.  STOP accepting new requests; return Unavailable for queued requests
 
3.  // Drain in-flight operations
    in_flight := client.pool.active_connections()
    AWAIT all_completed(in_flight, timeout = grace_period * 0.7)
 
4.  // Force-close remaining
    remaining := client.pool.active_connections()
    FOR EACH c IN remaining:
        c.send_goaway()
        c.close(timeout = grace_period * 0.2)
 
5.  // Flush telemetry buffers
    client.telemetry.flush(timeout = grace_period * 0.1)
 
6.  // Release resources
    client.pool.destroy()
    client.state := CLOSED
 
7.  EMIT event: SDK_SHUTDOWN { in_flight_drained, force_closed, flush_success }

ALGORITHM: RetryInterceptor(request: Request, next: Next, policy: RetryPolicy) → Result<Response, SdkError>
 
1.  attempt := 0
2.  last_error := NULL
 
3.  WHILE attempt < policy.max_attempts:
        // Ensure idempotency key is set for retries
        IF attempt > 0 AND request.idempotency_key IS NULL:
            IF request.is_mutating → RETURN Err(last_error)  // Cannot safely retry
            // Read-only requests are inherently idempotent
 
        TRY:
            response := next(request)
            IF response.is_success OR NOT policy.retry_on(response.status):
                RETURN Ok(response)
            last_error := classify_error(response)
        CATCH error:
            last_error := classify_error(error)
            IF NOT policy.retry_on(last_error.class):
                RETURN Err(last_error)
 
        attempt += 1
 
        // Compute backoff
        delay := compute_backoff(attempt, policy)
        EMIT metric: sdk.retry { attempt, delay_ms, error_class }
        SLEEP(delay)
 
4.  RETURN Err(last_error.with_context("Exhausted retry budget", attempts = attempt))
 
FUNCTION compute_backoff(attempt: int, policy: RetryPolicy) → Duration:
    base := policy.initial_backoff * (policy.multiplier ^ (attempt - 1))
    capped := MIN(base, policy.max_backoff)
    jittered := capped * RANDOM(0.5, 1.0)  // Half-jitter
    RETURN jittered

ALGORITHM: RateLimitInterceptor(request: Request, next: Next, bucket: TokenBucket) → Result<Response, SdkError>
 
1.  cost := estimate_token_cost(request)  // Based on request size, operation class
 
2.  IF bucket.try_acquire(cost):
        RETURN next(request)
 
3.  // Check server-provided retry-after header from previous 429
    IF bucket.retry_after IS SET AND now() < bucket.retry_after:
        wait_time := bucket.retry_after - now()
    ELSE:
        wait_time := bucket.time_to_refill(cost)
 
4.  IF wait_time > request.deadline.remaining():
        RETURN Err(RateLimited("Insufficient budget within deadline"))
 
5.  SLEEP(wait_time)
6.  IF bucket.try_acquire(cost):
        RETURN next(request)
7.  ELSE:
        RETURN Err(RateLimited("Rate limit exceeded after wait"))

ALGORITHM: AuthInterceptor(request: Request, next: Next, provider: AuthProvider) → Result<Response, SdkError>
 
1.  token := provider.get_cached_token()
 
2.  IF token IS NULL OR token.expires_at < now() + REFRESH_BUFFER:
        token := provider.refresh_token()
        IF token IS Err → RETURN Err(AuthFailure("Token refresh failed"))
 
3.  request.headers["Authorization"] := "Bearer " + token.value
4.  request.headers["X-SDK-Client-Id"] := provider.client_id
 
5.  response := next(request)
 
6.  IF response.status = 401 (Unauthorized):
        // Token may have been revoked; force refresh
        provider.invalidate_cache()
        token := provider.refresh_token()
        IF token IS Err → RETURN Err(AuthFailure("Re-auth failed"))
        request.headers["Authorization"] := "Bearer " + token.value
        response := next(request)
 
7.  RETURN response

ALGORITHM: ParallelToolInvoke(
    registry: ToolRegistry,
    calls: List<ToolCall>,
    max_concurrency: int,
    deadline: Duration
) → Result<List<ToolOutput>, SdkError>
 
1.  CREATE scope S with deadline and cancellation_token
 
2.  semaphore := Semaphore(max_concurrency)
    results := ConcurrentMap<int, Result<ToolOutput, SdkError>>()
 
3.  FOR EACH (index, call) IN enumerate(calls):
        SPAWN task IN scope S:
            ACQUIRE semaphore (or AWAIT within deadline)
            TRY:
                output := registry.invoke(call.tool_id, call.input, InvokeOptions {
                    deadline = deadline,
                    cancellation = S.token
                })
                results[index] := output
            CATCH error:
                results[index] := Err(error)
                IF error.class IN {AuthFailure, InvalidRequest}:
                    S.cancel("Non-retryable tool failure")
            FINALLY:
                RELEASE semaphore
 
4.  AWAIT scope S completion
 
5.  // Collect results in original order
    ordered := [results[i] FOR i IN 0..len(calls)]
 
6.  // Determine aggregate outcome
    failures := [r FOR r IN ordered WHERE r IS Err]
    IF len(failures) > 0 AND policy = FAIL_FAST:
        RETURN Err(AggregateError(failures))
 
7.  RETURN Ok(ordered)

ALGORITHM: ProcessAgentStream(
    stream: AsyncStream<AgentChunk>,
    handler: ChunkHandler,
    budget: TokenBudget
) → Result<AgentResponse, SdkError>
 
1.  accumulated_tokens := 0
    assembled_response := ResponseAssembler()
 
2.  FOR EACH chunk IN stream:
        // Backpressure: pause consumption if handler is slow
        AWAIT handler.ready()
 
        // Token budget enforcement
        accumulated_tokens += chunk.token_count
        IF accumulated_tokens > budget.max_output_tokens:
            stream.cancel("Output token budget exceeded")
            BREAK
 
        // Deliver to handler
        handler.on_chunk(chunk)
        assembled_response.append(chunk)
 
        // Check for tool-call chunks requiring interleaved execution
        IF chunk.type = TOOL_CALL_REQUEST:
            tool_result := AWAIT execute_tool_call(chunk.tool_call)
            stream.inject_context(tool_result)  // Feed back into stream
 
3.  IF stream.error IS NOT NULL:
        RETURN Err(stream.error)
 
4.  RETURN Ok(assembled_response.finalize())

ALGORITHM: SyncOnReconnect(offline_store: OfflineStore, platform: AgentClient)
 
1.  DETECT connectivity restored (via health check probe)
 
2.  // Phase 1: Upload buffered telemetry
    telemetry_batch := offline_store.drain_telemetry_buffer()
    platform.telemetry.batch_upload(telemetry_batch)
 
3.  // Phase 2: Replay write-ahead log for memory writes
    wal_entries := offline_store.read_wal(order = CHRONOLOGICAL)
    FOR EACH entry IN wal_entries:
        result := platform.memory.write(entry.item)
        IF result IS Ok:
            offline_store.wal.mark_committed(entry.id)
        ELSE IF result.error.class = ConflictError:
            // Conflict resolution
            resolved := conflict_resolver.resolve(entry, result.server_version)
            platform.memory.write(resolved)
            offline_store.wal.mark_committed(entry.id)
        ELSE:
            offline_store.wal.mark_retry(entry.id)
 
4.  // Phase 3: Refresh cached tool schemas
    stale_schemas := offline_store.tool_cache.get_stale(threshold = schema_ttl)
    FOR EACH schema IN stale_schemas:
        updated := platform.tools.get_schema(schema.tool_id)
        IF updated.version > schema.version:
            offline_store.tool_cache.update(updated)
 
5.  // Phase 4: Sync local memory replica
    last_sync := offline_store.memory_replica.last_sync_timestamp
    delta := platform.memory.query(MemoryQuery { modified_after = last_sync })
    offline_store.memory_replica.apply_delta(delta)
 
6.  EMIT event: SYNC_COMPLETE {
        wal_committed, wal_conflicted, schemas_refreshed, memory_items_synced
    }

TYPE LocalInferenceConfig:
    model_path: FilePath              // GGUF, ONNX, CoreML, TFLite
    model_format: ModelFormat         // GGUF | ONNX | COREML | TFLITE
    context_window: int               // Typically 2048–8192 for edge models
    max_output_tokens: int
    quantization: QuantizationLevel   // Q4_K_M, Q8_0, FP16, etc.
    device: InferenceDevice           // CPU | GPU | NPU | ANE
    batch_size: int
    temperature: float

ALGORITHM: VersionNegotiation(client_version: Version, server_capabilities: ServerCapabilities) → NegotiationResult
 
1.  IF server_capabilities.min_sdk_version > client_version:
        RETURN NegotiationResult.UNSUPPORTED(
            "Server requires SDK >= " + server_capabilities.min_sdk_version
        )
 
2.  IF client_version.MAJOR != server_capabilities.current_major:
        IF server_capabilities.compat_shims.contains(client_version.MAJOR):
            RETURN NegotiationResult.DEGRADED(
                active_shim = server_capabilities.compat_shims[client_version.MAJOR],
                warnings = ["Operating in compatibility mode"]
            )
        ELSE:
            RETURN NegotiationResult.UNSUPPORTED(...)
 
3.  // Negotiate feature flags
    agreed_features := INTERSECT(
        client_version.supported_features,
        server_capabilities.available_features
    )
 
4.  RETURN NegotiationResult.FULL(
        protocol_version = server_capabilities.current_protocol,
        features = agreed_features,
        deprecated_methods = server_capabilities.deprecations.filter(
            affects = client_version
        )
    )

TYPE DeprecationInfo:
    method: MethodSignature
    deprecated_since: Version
    removal_target: Version
    replacement: MethodSignature?
    migration_guide_url: URI
    reason: string

TYPE SchemaIR:
    messages: List<MessageDef>
    services: List<ServiceDef>
    enums: List<EnumDef>
    errors: List<ErrorDef>
    extensions: Map<string, Any>
 
TYPE MessageDef:
    name: QualifiedName
    fields: List<FieldDef>
    oneofs: List<OneofDef>
    nested: List<MessageDef>
    options: MessageOptions       // e.g., deprecated, experimental
 
TYPE FieldDef:
    name: string
    number: int
    type: FieldType               // Scalar, Message, Enum, Map, Repeated
    optional: bool
    default: Value?
    constraints: List<Constraint>  // min, max, pattern, custom validators
    documentation: DocString
    deprecation: DeprecationInfo?
 
TYPE ServiceDef:
    name: QualifiedName
    methods: List<MethodDef>
    version: SemanticVersion
 
TYPE MethodDef:
    name: string
    input: MessageRef
    output: MessageRef
    streaming: StreamingMode      // UNARY | SERVER_STREAM | CLIENT_STREAM | BIDI_STREAM
    idempotency: IdempotencyLevel // UNKNOWN | IDEMPOTENT | NO_SIDE_EFFECTS
    deadline_class: DeadlineClass // FAST (<100ms) | NORMAL (<5s) | SLOW (<60s) | LONG_RUNNING
    auth_required: bool
    approval_gated: bool          // Requires human approval for execution

ALGORITHM: GenerateSDK(
    schemas: List<SchemaSource>,
    targets: List<LanguageTarget>,
    config: CodegenConfig
) → Map<LanguageTarget, GeneratedModule>
 
1.  // Stage 1: Parse and validate
    ir := SchemaIR.empty()
    FOR EACH source IN schemas:
        parsed := MATCH source.format:
            PROTOBUF  → parse_proto(source.path)
            JSON_SCHEMA → parse_jsonschema(source.path)
            OPENAPI    → parse_openapi(source.path)
            MCP_MANIFEST → parse_mcp(source.path)
        validate_no_conflicts(ir, parsed)
        ir.merge(parsed)
 
2.  // Stage 2: Transform and augment
    FOR EACH message IN ir.messages:
        message.add_builder_pattern()
        message.add_validation_methods(from = message.fields.constraints)
        message.add_equality_and_hash()
        message.add_serialization_annotations()
        IF message.has_deprecated_fields():
            message.add_deprecation_warnings()
 
    FOR EACH service IN ir.services:
        FOR EACH method IN service.methods:
            method.add_retry_annotation(from = method.idempotency)
            method.add_deadline_default(from = method.deadline_class)
            method.add_interceptor_hooks()
 
3.  // Stage 3: Emit per target language
    results := Map.empty()
    FOR EACH target IN targets:
        emitter := LanguageEmitter.create(target, config)
        raw_code := emitter.emit(ir)
        
        // Stage 4: Post-process
        formatted := target.formatter.format(raw_code)
        linted := target.linter.check(formatted)
        IF linted.has_errors():
            RAISE CodegenError("Generated code has lint errors", linted.errors)
 
        // Generate snapshot tests
        tests := generate_serialization_round_trip_tests(ir, target)
        tests += generate_validation_tests(ir, target)
        tests += generate_builder_tests(ir, target)
 
        results[target] := GeneratedModule {
            source_files = formatted,
            test_files = tests,
            manifest = generate_package_manifest(target, config.version)
        }
 
4.  // Cross-language conformance: verify identical serialization output
    reference_vectors := generate_conformance_vectors(ir)
    FOR EACH target IN targets:
        verify_conformance(results[target], reference_vectors)
 
5.  RETURN results

TYPE MockServer:
    FUNC start(config: MockServerConfig) → MockServerHandle
    FUNC stop() → void
 
    // Programming interface
    FUNC when(matcher: RequestMatcher) → ResponseStub
    FUNC verify(matcher: RequestMatcher, times: InvocationCount) → VerificationResult
    FUNC record() → RecordingSession
    FUNC replay(session: RecordedSession) → void
 
TYPE RequestMatcher:
    method: MethodPattern          // Exact, regex, wildcard
    body: BodyMatcher              // JSON path assertions, schema match
    headers: HeaderMatcher
    metadata: MetadataMatcher
 
TYPE ResponseStub:
    FUNC respond_with(response: Response) → void
    FUNC respond_with_sequence(responses: List<Response>) → void
    FUNC respond_with_stream(chunks: List<Chunk>, interval: Duration) → void
    FUNC respond_with_error(error: SdkError) → void
    FUNC respond_with_delay(response: Response, delay: Duration) → void
    FUNC respond_with_function(handler: (Request) → Response) → void

ALGORITHM: DeterministicReplayTest(
    recorded_session: RecordedSession,
    sdk_under_test: AgentClient
) → TestResult
 
1.  // Phase 1: Load recorded interactions
    interactions := recorded_session.load()
    // Each interaction: (timestamp, request, response, metadata)
 
2.  // Phase 2: Configure mock server with recorded responses
    mock := MockServer.start(config = { transport = recorded_session.transport })
    FOR EACH (i, interaction) IN enumerate(interactions):
        mock.when(
            RequestMatcher.exact(interaction.request)
        ).respond_with(interaction.response)
 
3.  // Phase 3: Execute SDK operations against mock
    sdk_under_test.connect(mock.endpoint)
    actual_results := []
 
    FOR EACH interaction IN interactions:
        actual := sdk_under_test.execute(interaction.request, interaction.deadline)
        actual_results.append(actual)
 
4.  // Phase 4: Verify behavioral equivalence
    mismatches := []
    FOR EACH (expected, actual) IN zip(interactions, actual_results):
        IF NOT deep_equals(expected.response, actual, config.comparison_rules):
            mismatches.append(Mismatch {
                index = i,
                expected = expected.response,
                actual = actual,
                diff = compute_diff(expected.response, actual)
            })
 
5.  // Phase 5: Verify side effects
    FOR EACH interaction IN interactions:
        mock.verify(
            RequestMatcher.exact(interaction.request),
            times = EXACTLY_ONCE
        )
 
6.  mock.stop()
    sdk_under_test.disconnect()
 
7.  IF mismatches IS EMPTY:
        RETURN TestResult.PASSED(interactions_verified = len(interactions))
    ELSE:
        RETURN TestResult.FAILED(mismatches)

ALGORITHM: RecordSession(client: AgentClient, operations: List<Operation>) → RecordedSession
 
1.  recorder := SessionRecorder.start()
    client.use(RecordingInterceptor(recorder))
    // RecordingInterceptor captures all request/response pairs with timestamps
 
2.  FOR EACH op IN operations:
        EXECUTE op via client
        // RecordingInterceptor automatically captures interaction
 
3.  session := recorder.finalize()
 
4.  // Sanitize sensitive data
    session.redact(patterns = [
        RedactPattern.BEARER_TOKEN,
        RedactPattern.API_KEY,
        RedactPattern.PII_FIELDS
    ])
 
5.  // Compute determinism hash
    session.hash := SHA256(canonical_serialize(session.interactions))
 
6.  session.save(path = config.recording_dir / session.id + ".session.json")
 
7.  RETURN session

TYPE BrowserSdkConfig:
    endpoint: URL                   // Must be HTTPS with valid CORS
    transport: FETCH | SSE | WEBSOCKET
    auth: BrowserAuthConfig         // Cookie, bearer token, or OAuth PKCE flow
    worker: boolean                 // Offload to Web Worker
    bundle_budget: ByteSize         // Max JS bundle size
    storage: LOCAL_STORAGE | INDEXED_DB | MEMORY
    wasm_modules: List<WasmModule>  // Optional WASM for local inference

// Tree-shakeable exports
export { AgentClient } from './core/client'
export { ToolRegistry } from './tools/registry'       // Optional
export { MemoryStore } from './memory/store'           // Optional
export { ContextBuilder } from './context/builder'     // Optional
// Consumer only imports what they use; bundler eliminates the rest

ALGORITHM: ServerlessSdkLifecycle(handler: RequestHandler) → LambdaHandler
 
1.  // INIT PHASE (cold start, runs once)
    GLOBAL client := AgentClient.create(ServerlessConfig {
        pool_size = 1,                    // Minimal pooling
        keepalive = true,                  // Reuse across invocations
        lazy_tool_discovery = true,        // Don't discover until needed
        telemetry_mode = BATCHED,          // Buffer and flush at end
        memory_mode = STATELESS            // No local memory persistence
    })
    GLOBAL initialized := false
 
2.  // HANDLER (runs per invocation)
    RETURN ASYNC FUNCTION(event, context):
        IF NOT initialized:
            AWAIT client.connect()
            initialized := true
 
        // Set invocation-scoped deadline from Lambda remaining time
        deadline := context.remaining_time - SAFETY_MARGIN(500ms)
 
        TRY:
            result := AWAIT handler(client, event, deadline)
            RETURN format_response(result)
        CATCH error:
            RETURN format_error(error)
        FINALLY:
            // Flush telemetry before freeze
            AWAIT client.telemetry.flush(timeout = MIN(deadline * 0.1, 1000ms))
            // Do NOT disconnect; connection persists across warm invocations

ALGORITHM: AirGappedSdkSetup(bundle: AirGapBundle) → AgentClient
 
1.  // Validate bundle integrity
    IF SHA256(bundle.content) != bundle.manifest.checksum:
        RAISE SecurityError("Bundle integrity check failed")
 
2.  // Extract components
    model_weights := bundle.extract("model/")
    tool_schemas := bundle.extract("tools/")
    memory_snapshot := bundle.extract("memory/")
    policy_config := bundle.extract("config/policy.json")
 
3.  // Initialize local-only SDK
    client := AgentClient.create(AirGapConfig {
        inference = LocalInferenceConfig {
            model_path = model_weights,
            device = detect_available_device()
        },
        tools = StaticToolRegistry(tool_schemas),
        memory = FileBackedMemoryStore(memory_snapshot),
        telemetry = FileExporter(path = "/var/log/agent/telemetry/"),
        connectivity = NEVER    // No network operations attempted
    })
 
4.  RETURN client

TYPE TelemetryConfig:
    traces: TraceConfig
    metrics: MetricsConfig
    logs: LogConfig
    provenance: ProvenanceConfig
    export: ExportConfig           // OTLP, stdout, file, custom
    sampling: SamplingConfig
    privacy: PrivacyConfig         // PII redaction, content filtering
    budget: TelemetryBudget        // Max overhead: CPU%, memory, bandwidth

[sdk.agent.execute]                              // Root span
  ├── [sdk.context.compile]                      // Context compilation
  │     ├── [sdk.context.retrieve]               // Evidence retrieval
  │     ├── [sdk.memory.summarize]               // Memory summarization
  │     └── [sdk.context.budget_allocate]         // Token budget allocation
  ├── [sdk.transport.request]                    // Wire-level request
  │     ├── [sdk.middleware.auth]                // Auth injection
  │     ├── [sdk.middleware.retry]               // Retry wrapper
  │     │     ├── [sdk.transport.attempt.1]      // First attempt
  │     │     └── [sdk.transport.attempt.2]      // Retry
  │     └── [sdk.middleware.circuit_breaker]     // Circuit breaker check
  ├── [sdk.tool.invoke]                          // Tool invocation (if streaming)
  └── [sdk.response.deserialize]                 // Response processing

ALGORITHM: AdaptiveSampler(config: SamplingConfig) → SamplingDecision
 
STATE:
    window: SlidingWindow(size = 60s)
    current_rate: float = config.initial_rate    // e.g., 1.0 = 100%
    overhead_ema: float = 0.0                    // Exponential moving average
 
FUNCTION should_sample(span: SpanContext) → bool:
    // Priority sampling: always sample errors and slow requests
    IF span.has_error OR span.duration > config.slow_threshold:
        RETURN true
 
    // Always sample if trace is already being sampled (maintain trace integrity)
    IF span.parent_sampled:
        RETURN true
 
    // Probabilistic sampling based on current rate
    IF RANDOM() < current_rate:
        RETURN true
 
    RETURN false
 
FUNCTION update_rate():
    // Called every adjustment_interval (e.g., 5 seconds)
    measured_overhead := measure_telemetry_cpu_percent()
    overhead_ema := 0.3 * measured_overhead + 0.7 * overhead_ema
 
    IF overhead_ema > config.max_overhead:
        current_rate := MAX(config.min_rate, current_rate * 0.8)
    ELSE IF overhead_ema < config.max_overhead * 0.5:
        current_rate := MIN(1.0, current_rate * 1.1)
 
    EMIT metric: sdk.telemetry.sample_rate { value = current_rate }

TYPE SdkDiagnosticReport:
    // Error context
    error: SdkError
    error_chain: List<SdkError>          // Causal chain
    request_summary: RequestSummary       // Redacted request details
 
    // Execution context
    trace_id: TraceId
    span_id: SpanId
    agent_loop_state: LoopState?         // Phase, iteration, step
 
    // Environmental context
    sdk_version: Version
    transport: TransportType
    connection_health: HealthStatus
    circuit_breaker_state: CircuitState
    pool_utilization: float
 
    // Timing breakdown
    timing: TimingBreakdown {
        total_ms: float,
        queue_wait_ms: float,
        connection_acquire_ms: float,
        serialization_ms: float,
        network_ms: float,
        deserialization_ms: float,
        middleware_ms: Map<InterceptorName, float>
    }
 
    // Suggested actions
    suggestions: List<DiagnosticSuggestion>

ALGORITHM: GenerateDiagnosticSuggestions(report: SdkDiagnosticReport) → List<DiagnosticSuggestion>
 
1.  suggestions := []
 
2.  IF report.error.class = DeadlineExceeded:
        IF report.timing.connection_acquire_ms > report.timing.total_ms * 0.5:
            suggestions.append("Increase connection pool size or reduce concurrency")
        IF report.timing.network_ms > report.timing.total_ms * 0.7:
            suggestions.append("Server latency is dominant; consider increasing deadline or optimizing server-side processing")
        IF report.timing.serialization_ms > 100:
            suggestions.append("Large request payload; consider compression or chunked transmission")
 
3.  IF report.error.class = RateLimited:
        suggestions.append("Reduce request rate or increase rate limit quota")
        suggestions.append("Enable request queuing with backpressure")
 
4.  IF report.circuit_breaker_state = OPEN:
        suggestions.append("Circuit breaker is open due to high failure rate; investigate server health")
        suggestions.append("Check server-side logs for trace_id: " + report.trace_id)
 
5.  IF report.pool_utilization > 0.9:
        suggestions.append("Connection pool near exhaustion; consider scaling pool_max_size")
 
6.  IF report.error_chain.length > 3:
        suggestions.append("Deep error chain suggests cascading failure; review retry policy")
 
7.  RETURN suggestions

TYPE SdkProfiler:
    FUNC start_profile(config: ProfileConfig) → ProfileSession
    FUNC stop_profile(session: ProfileSession) → ProfileReport
 
TYPE ProfileReport:
    duration: Duration
    operations: List<ProfiledOperation>
    hotspots: List<Hotspot>              // Sorted by cumulative time
    allocation_summary: AllocationSummary
    network_summary: NetworkSummary
    token_usage_summary: TokenUsageSummary
 
    FUNC export(format: FLAMEGRAPH | CHROME_TRACE | PPROF) → bytes

ALGORITHM: RedactTelemetry(item: TelemetryItem, policy: PrivacyPolicy) → TelemetryItem
 
1.  // Redact known PII patterns
    FOR EACH field IN item.string_fields():
        FOR EACH pattern IN policy.pii_patterns:
            field.value := regex_replace(field.value, pattern, "[REDACTED]")
 
2.  // Remove content fields if policy forbids content export
    IF policy.export_content = false:
        item.remove_fields(["request.body", "response.body", "context.content"])
 
3.  // Hash user identifiers for correlation without exposure
    IF policy.hash_user_ids:
        item.user_id := HMAC_SHA256(policy.salt, item.user_id)
 
4.  // Enforce attribute allowlist
    FOR EACH attr IN item.attributes:
        IF attr.key NOT IN policy.allowed_attributes:
            item.remove_attribute(attr.key)
 
5.  RETURN item