Source Code Reading: FunctionRuntime — Unified Multi-Language Sandbox Interface

“

Series: S9 Source Code Reading · Article 14 | Level: Advanced | Reading Time: 25 min

#TL;DR

FunctionRuntimeService is the function execution engine within coomia-dip's Reasoning & Decision layer (Reasoning & Decision Layer), providing unified registration, invocation, and lifecycle management for Python, TypeScript, and Groovy. It manages function definitions through FunctionRegistry and implements sandboxed execution through FunctionExecutor, supporting five function types (QUERY/DERIVED/ACTION/AGGREGATION/TIMESERIES) and three execution modes (SERVERLESS/DEPLOYED/PIPELINE). This article dissects the Pydantic request model to Proto contract mapping strategy, the Registry/Executor dual-layer architecture, WorldContext injection into functions, and async invocation status tracking design.

#Table of Contents

1. Overall Architecture: Registry + Executor Dual-Layer Design
2. Proto Contract: FunctionDefinition's Five Types and Three Modes
3. FunctionRuntimeServicer: gRPC Adaptation Layer
4. Registration Flow: Proto to Domain Conversion Chain
5. Function Invocation: WorldContext Injection and Sandbox Execution
6. Exception Layering: Four-Level Error Handling
7. Async Invocation and Status Tracking
8. Code Binding Generation: GenerateCodeBindings
9. Function Update: Config Merge Strategy
10. InvocationMetrics: Performance Observability
11. Key Takeaways

#1. Overall Architecture: Registry + Executor Dual-Layer Design

FunctionRuntime employs a classic registry + executor architecture, completely separating function "definition management" from "runtime execution":

intelligence-Layer/src/reasoning_decision_plane/
├── function_runtime/
│   ├── registry.py               # FunctionRegistry -- function registry
│   ├── executor.py               # FunctionExecutor -- sandbox executor
│   └── models.py                 # Domain models (Pydantic v2)
├── api/grpc/
│   └── function_runtime_servicer.py  # gRPC Servicer adapter
├── common/
│   ├── context.py                # WorldContext
│   └── exceptions.py             # Unified exception hierarchy

Key design decision: FunctionRuntimeServicer accepts FunctionRegistry and FunctionExecutor through constructor injection, both providing factory functions as default instance accessors:

class FunctionRuntimeServicer:
    def __init__(
        self,
        registry: FunctionRegistry | None = None,
        executor: FunctionExecutor | None = None,
    ) -> None:
        self._registry = registry or get_function_registry()
        self._executor = executor or get_function_executor()

This "injectable with defaults" pattern is particularly useful in testing -- unit tests can inject mocks while integration tests use real instances.

#2. Proto Contract: FunctionDefinition's Five Types and Three Modes

The Proto's FunctionDefinition carries complete function metadata across five types mapping to Palantir Foundry's function taxonomy:

enum FunctionType {
  FUNCTION_TYPE_QUERY = 1;           // Read-only queries
  FUNCTION_TYPE_DERIVED = 2;         // Derived property computation
  FUNCTION_TYPE_ACTION = 3;          // Mutating operations
  FUNCTION_TYPE_AGGREGATION = 4;     // ObjectSet aggregation
  FUNCTION_TYPE_TIMESERIES = 5;      // Time series processing
}

• QUERY: Read-only functions whose results can be safely cached
• DERIVED: Derived property calculations, integrated with DerivedPropertyService
• ACTION: Mutating operations invoked by ActionEngine
• AGGREGATION: Aggregation computations over ObjectSets
• TIMESERIES: Time series data processing

Three execution modes balance resource efficiency and startup latency: SERVERLESS for low-frequency calls, DEPLOYED for high-frequency hot paths, PIPELINE for batch processing scenarios.

#3. FunctionRuntimeServicer: gRPC Adaptation Layer

The Servicer's core responsibility is converting between Proto request/response models and internal Domain models. It uses Pydantic BaseModel as an intermediate layer:

class RegisterFunctionRequest(BaseModel):
    function_id: str = ""
    function_name: str = ""
    function_type: str = "QUERY"
    language: str = "PYTHON"
    code: str = ""
    entrypoint: str = "main"
    description: str = ""
    timeout_seconds: int = 30
    max_memory_mb: int = 256

Default value strategy: All fields have sensible defaults. function_type defaults to "QUERY" (the safest read-only type), language defaults to "PYTHON" (the platform's primary language), entrypoint defaults to "main" (Python convention). This means a minimal registration request only needs function_name and code.

FunctionResponse carries both success and failure cases through error_code / error_message fields -- when error_code is non-empty, the operation failed. This "unified response" pattern avoids throwing gRPC-level exceptions and lets clients handle success and failure uniformly.

#4. Registration Flow: Proto to Domain Conversion Chain

The register_function method demonstrates the complete Proto-to-Domain conversion chain:

async def register_function(self, request: RegisterFunctionRequest) -> FunctionResponse:
    try:
        # 1. Parse language enum (graceful fallback to PYTHON)
        try:
            language = FunctionLanguage(request.language.lower())
        except ValueError:
            language = FunctionLanguage.PYTHON

        # 2. Parse function type (graceful fallback to QUERY)
        try:
            func_type = FunctionType(request.function_type.lower())
        except ValueError:
            func_type = FunctionType.QUERY

        # 3. Build FunctionConfig
        config = FunctionConfig(
            timeout_seconds=request.timeout_seconds or 30,
            max_memory_mb=request.max_memory_mb or 256,
        )

        # 4. Call Registry to register
        func_def = await self._registry.register(
            function_id=request.function_id or None,
            function_name=request.function_name,
            type=func_type, language=language,
            code=request.code, entrypoint=request.entrypoint or "main",
            config=config, description=request.description,
        )

Graceful degradation pattern is the key design point: enum parsing failures don't throw exceptions but degrade to the safest default values (PYTHON/QUERY). This keeps the system robust against non-conforming client inputs.

#5. Function Invocation: WorldContext Injection and Sandbox Execution

invoke_function is the most critical method, demonstrating the complete function invocation flow:

async def invoke_function(self, request: InvokeFunctionRequest) -> InvokeFunctionResponse:
    try:
        world_context: WorldContext | None = None
        if request.world_id:
            world_context = WorldContext(world_id=request.world_id)

        result = await self._executor.invoke(
            function_id=request.function_id,
            arguments=request.arguments,
            world_context=world_context,
            timeout_override=request.timeout_override or None,
        )

WorldContext injection: When the caller provides a world_id, a WorldContext object is constructed and injected into the executor, allowing functions to be aware of the current world context -- for example, which Nessie branch to query against.

The Proto-level FunctionArgument supports seven value types through oneof, including ObjectSetRef and ObjectRef that bridge the function system with the ontology system -- functions can directly receive ontology objects or object sets as parameters.

#6. Exception Layering: Four-Level Error Handling

FunctionRuntime defines four specialized exceptions forming a clear exception handling hierarchy: NOT_FOUND, TIMEOUT, EXECUTION_ERROR, and INTERNAL_ERROR, from most specific to most generic.

except FunctionNotFoundError as e:
    return InvokeFunctionResponse(error_code="NOT_FOUND", error_message=str(e))
except FunctionTimeoutError as e:
    return InvokeFunctionResponse(error_code="TIMEOUT", error_message=str(e))
except FunctionExecutionError as e:
    return InvokeFunctionResponse(
        error_code="EXECUTION_ERROR",
        error_message=e.error_message,
        stack_trace=e.stack_trace or "",
    )
except Exception as e:
    return InvokeFunctionResponse(error_code="INTERNAL_ERROR", error_message=str(e))

FunctionExecutionError carries an additional stack_trace field to help developers locate errors within the function. Even on exception paths, responses generate valid invocation_id values (using UUID hex prefix) so clients can correlate logs for troubleshooting.

#7. Async Invocation and Status Tracking

The Proto defines separate async invocation and status query interfaces:

service FunctionRuntimeService {
  rpc InvokeFunction(InvokeFunctionRequest) returns (InvokeFunctionResponse);
  rpc InvokeFunctionAsync(InvokeFunctionRequest) returns (AsyncInvocationResponse);
  rpc GetInvocationStatus(GetInvocationStatusRequest) returns (InvocationStatus);
}

Async invocation returns AsyncInvocationResponse with just the invocation_id and initial state (typically PENDING). Clients poll GetInvocationStatus for progress and final results. Seven invocation states cover the complete async function lifecycle: PENDING, RUNNING, COMPLETED, FAILED, TIMEOUT, CANCELLED.

#8. Code Binding Generation: GenerateCodeBindings

GenerateCodeBindings is a clever developer experience feature that generates type-safe code bindings based on specified ontology types and target language:

message GenerateBindingsRequest {
  repeated string ontology_types = 2;
  FunctionLanguage target_language = 3;
}

message CodeBindingsResponse {
  map<string, string> bindings = 1;       // type_name -> generated code
}

For example, given the Employee ontology type and Python language, it generates a Pydantic model class containing all properties and relations. The bindings map stores type name to generated code text pairs.

#9. Function Update: Config Merge Strategy

The update_function method demonstrates a carefully designed partial update strategy:

async def update_function(self, request: UpdateFunctionRequest) -> FunctionResponse:
    config: FunctionConfig | None = None
    if request.timeout_seconds is not None or request.max_memory_mb is not None:
        existing = await self._registry.get_or_none(request.function_id)
        if existing:
            config = FunctionConfig(
                timeout_seconds=request.timeout_seconds
                if request.timeout_seconds is not None
                else existing.config.timeout_seconds,
                max_memory_mb=request.max_memory_mb
                if request.max_memory_mb is not None
                else existing.config.max_memory_mb,
                mode=existing.config.mode,
                allowed_apis=existing.config.allowed_apis,
                env_vars=existing.config.env_vars,
            )

Merge rather than overwrite: When an update request only modifies timeout_seconds, all other configuration fields are preserved from the existing function definition. The request model uses int | None to distinguish between "not provided" and "set to 0" semantics.

#10. InvocationMetrics: Performance Observability

Every function invocation returns detailed execution metrics across four dimensions: execution_time_ms for SLA monitoring, memory_used_mb for resource quota management, cpu_time_ms for billing and capacity planning, and api_calls for detecting functions that over-call external services.

#11. Key Takeaways

1. Registry/Executor separation: Decoupling function definition management from runtime execution allows independent scaling and testing.
2. Graceful degradation pattern: Enum parsing failures degrade to safest defaults rather than throwing exceptions, improving robustness.
3. Unified response pattern: Success and failure share the same Response type, distinguished by error_code, simplifying client-side handling.
4. Config merge updates: Partial updates preserve unmodified config fields, avoiding the "update causes data loss" API anti-pattern.
5. Seven argument types: FunctionArgument's oneof value covers the full type spectrum from primitives to ontology object references.
6. Four-dimensional execution metrics: Time/memory/CPU/API-calls combination provides comprehensive performance observability.

#Next Article

S9-15: DerivedPropertyService -- Dependency DAG and Cascade, where we dive into Reasoning & Decision Layer's derived property engine to understand four computation modes and the three-axis version model.

Tags: #coomia-dip #source-code-reading #function-runtime #multi-language #sandbox #grpc #Layer-d