diff --git a/dsl/product/compute_fabric.dag b/dsl/product/compute_fabric.dag index 80a4809a6b1..12bd9bfeb3c 100644 --- a/dsl/product/compute_fabric.dag +++ b/dsl/product/compute_fabric.dag @@ -33,6 +33,7 @@ import std.measure { hardware_thread_count, hardware_thread_count_value, measure_scale_fraction_ceil, + measure_scale_fraction_floor, hertz, watt, Bandwidth, @@ -312,7 +313,7 @@ type ComputeSupplyFacts { type ProviderConstraint = HostJobserverFifo { fifo_path: NonEmptyStr, config_ref: CtrlJobserverConfig } | SharedHomeRoot { path: NonEmptyStr } - | MaxConcurrentRunners { cap: Int } + | MaxConcurrentHeavyComputeLanes { cap: Int } | OomBehavior { signal: OomSignalClass } type AvailabilityWindow { open: LogicalTime, close: LogicalTime? } @@ -799,6 +800,240 @@ fn satisfies(offer: ComputeOffer, demand: WorkDemand) -> ComputeLeaseEligibility } } +type ControlPlaneAgent + = ClaudeCode + | Codex + | Cursor + +type SessionSubject { + agent: ControlPlaneAgent +} + +type ComputeLocality + = FulfillableRemotely + | MustRunLocal + +type ComputeWorkload + = ExecCommand { command: NonEmptyStr } + | GhaRunner { repo: NonEmptyStr, labels: List } + +type ComputeRequest { + requester: SessionSubject + workload: ComputeWorkload + needs: ResourceEnvelope + locality: ComputeLocality +} + +type ExecutorTarget + = BuildBuddyRemote { endpoint: NonEmptyStr } + | LeasedComputeLane { lease_id: HeavyComputeLeaseId, host: HostIdentity } + +type RemoteExecutorHandle { + provider: ProviderIdentity + target: ExecutorTarget +} + +type FulfillmentRejection + = LocalCapacityExhausted { lane_count: Nat, held: Nat } + | NeedsUnsatisfiable { reason: MissingDemandFact } + | DuplicateLeaseId { lease_id: HeavyComputeLeaseId } + +type Fulfillment + = Fulfilled { executor: RemoteExecutorHandle } + | FulfillmentQueued { lane_count: Nat, held: Nat } + | FulfillmentRejected { reason: FulfillmentRejection } + +type FulfillmentOutcome { + fulfillment: Fulfillment + ledger: LeaseLedger +} + +type HeavyComputeLaneSource + = GithubCiRunner + | OnDemandSessionLease + +type HeavyComputeLeaseId = NonEmptyStr where brand("HeavyComputeLeaseId") + +type HeavyComputeLease { + lease_id: HeavyComputeLeaseId + source: HeavyComputeLaneSource + demand: WorkDemand +} + +type LeaseLedger { + held: List +} + +data empty_lease_ledger: LeaseLedger = LeaseLedger { held: [] } + +type HeavyComputePool { + lane_envelope: ResourceEnvelope + lane_count: Nat +} + +fn heavy_compute_lane_memory(pool: HeavyComputePool) -> ByteSize { + demand_envelope_memory(envelope: pool.lane_envelope) +} + +fn heavy_compute_pool_bytes(pool: HeavyComputePool) -> ByteSize { + measure_scale_fraction_floor(m: heavy_compute_lane_memory(pool: pool), num: pool.lane_count, den: 1) +} + +fn ledger_held_count(ledger: LeaseLedger) -> Nat { + fold(ledger.held, init: 0, f: (acc, _l) => acc + 1) +} + +fn ledger_holds_source(ledger: LeaseLedger, source: HeavyComputeLaneSource) -> Bool { + fold( + ledger.held, + init: false, + f: (acc, l) => acc || l.source == source + ) +} + +fn ledger_has_free_lane(pool: HeavyComputePool, ledger: LeaseLedger) -> Bool { + ledger_held_count(ledger: ledger) < pool.lane_count +} + +fn ledger_has_lease_id(ledger: LeaseLedger, lease_id: HeavyComputeLeaseId) -> Bool { + fold(ledger.held, init: false, f: (acc, l) => acc || l.lease_id == lease_id) +} + +fn ledger_conserves(pool: HeavyComputePool, ledger: LeaseLedger) -> Bool { + !(pool.lane_count < ledger_held_count(ledger: ledger)) +} + +type LeaseGrantOutcome + = LeaseGranted { ledger: LeaseLedger, granted: HeavyComputeLease } + | LeaseRejectedIneligible { reason: MissingDemandFact } + | LeaseRejectedPoolExhausted { lane_count: Nat, held: Nat } + | LeaseRejectedDuplicateId { lease_id: HeavyComputeLeaseId } + +fn lease_grant( + pool: HeavyComputePool, + ledger: LeaseLedger, + offer: ComputeOffer, + lease_id: HeavyComputeLeaseId, + source: HeavyComputeLaneSource, + demand: WorkDemand +) -> LeaseGrantOutcome { + match satisfies(offer: offer, demand: demand) { + Rejected { reason: r } => LeaseRejectedIneligible { reason: r } + Eligible { witness: _w } => + if ledger_has_lease_id(ledger: ledger, lease_id: lease_id) { + LeaseRejectedDuplicateId { lease_id: lease_id } + } else if ledger_has_free_lane(pool: pool, ledger: ledger) { + let lease = HeavyComputeLease { + lease_id: lease_id, + source: source, + demand: demand, + } + LeaseGranted { + ledger: LeaseLedger { held: concat(ledger.held, [lease]) }, + granted: lease, + } + } else { + LeaseRejectedPoolExhausted { + lane_count: pool.lane_count, + held: ledger_held_count(ledger: ledger), + } + } + } +} + +fn lease_release(ledger: LeaseLedger, lease_id: HeavyComputeLeaseId) -> LeaseLedger { + LeaseLedger { held: filter(ledger.held, l => l.lease_id != lease_id) } +} + +type ComputeFabric { + pool: HeavyComputePool + remote_executor: RemoteExecutorHandle + local_offer: ComputeOffer +} + +fn workload_lease_source(workload: ComputeWorkload) -> HeavyComputeLaneSource { + match workload { + ExecCommand { command: _ } => OnDemandSessionLease + GhaRunner { repo: _, labels: _ } => GithubCiRunner + } +} + +fn request_to_work_demand(request: ComputeRequest) -> WorkDemand { + WorkDemand { + resources: request.needs, + os: none, + isolation: IsolationRequirement { boundary: SharedHostHome }, + toolchains: [], + parallelism: SingleWorkItem, + data_locality: [], + effects: [], + input_envelope: EnvelopeUnknown, + } +} + +fn fulfill( + fabric: ComputeFabric, + ledger: LeaseLedger, + request: ComputeRequest, + lease_id: HeavyComputeLeaseId +) -> FulfillmentOutcome { + match request.locality { + FulfillableRemotely => + FulfillmentOutcome { + fulfillment: Fulfilled { executor: fabric.remote_executor }, + ledger: ledger, + } + MustRunLocal => + match lease_grant( + pool: fabric.pool, + ledger: ledger, + offer: fabric.local_offer, + lease_id: lease_id, + source: workload_lease_source(workload: request.workload), + demand: request_to_work_demand(request: request) + ) { + LeaseGranted { ledger: granted_ledger, granted: g } => + FulfillmentOutcome { + fulfillment: Fulfilled { + executor: RemoteExecutorHandle { + provider: fabric.local_offer.provider, + target: LeasedComputeLane { + lease_id: g.lease_id, + host: fabric.local_offer.supply.physical.identity, + }, + }, + }, + ledger: granted_ledger, + } + LeaseRejectedPoolExhausted { lane_count: lc, held: h } => + match request.workload { + GhaRunner { repo: _, labels: _ } => + FulfillmentOutcome { + fulfillment: FulfillmentQueued { lane_count: lc, held: h }, + ledger: ledger, + } + ExecCommand { command: _ } => + FulfillmentOutcome { + fulfillment: FulfillmentRejected { + reason: LocalCapacityExhausted { lane_count: lc, held: h }, + }, + ledger: ledger, + } + } + LeaseRejectedIneligible { reason: r } => + FulfillmentOutcome { + fulfillment: FulfillmentRejected { reason: NeedsUnsatisfiable { reason: r } }, + ledger: ledger, + } + LeaseRejectedDuplicateId { lease_id: dup } => + FulfillmentOutcome { + fulfillment: FulfillmentRejected { reason: DuplicateLeaseId { lease_id: dup } }, + ledger: ledger, + } + } + } +} + type WorkUnitId = NonEmptyStr where brand("WorkUnitId") type WorkUnit { @@ -1357,3 +1592,368 @@ data gunbc_ci_corpus_envelope_ceiling_scaffold: Disposition = Scaffold { field: WholeDeclaration } } + +data heavy_compute_lane_work_demand: WorkDemand = WorkDemand { + resources: ResourceEnvelope { + cpu: Present { + value: CpuRequirement { min_threads: hardware_thread_count(16), architecture: none }, + }, + gpu: none, + memory: Present { value: MemoryRequirement { min_bytes: byte_size(34359738368) } }, + storage: none, + network: none, + }, + os: none, + isolation: IsolationRequirement { boundary: SharedHostHome }, + toolchains: [], + parallelism: SingleWorkItem, + data_locality: [], + effects: [], + input_envelope: EnvelopeUnknown, +} + +data example_heavy_compute_pool: HeavyComputePool = HeavyComputePool { + lane_envelope: heavy_compute_lane_work_demand.resources, + lane_count: 2, +} + +data control_plane_agent_vendor_scaffold: Disposition = Scaffold { + dissolves_to: SingleAuthority, + bind: DeclarationRef { + module_path: "extdeps.vendor", + decl_name: "Vendor", + field: WholeDeclaration + } +} + +data heavy_compute_pool_lane_count_projection_scaffold: Disposition = Scaffold { + dissolves_to: SingleAuthority, + bind: DeclarationRef { + module_path: "gunbc.fleet_host_budget", + decl_name: "fleet_host_plan_for_offer", + field: WholeDeclaration + } +} + +data buildbuddy_remote_executor_vendor_scaffold: Disposition = Scaffold { + dissolves_to: SingleAuthority, + bind: DeclarationRef { + module_path: "extdeps.vendor", + decl_name: "Vendor", + field: WholeDeclaration + } +} + +data example_session_subject: SessionSubject = SessionSubject { agent: ClaudeCode } + +data example_remote_executor: RemoteExecutorHandle = RemoteExecutorHandle { + provider: "buildbuddy-remote", + target: BuildBuddyRemote { endpoint: "grpcs://remote.buildbuddy.io" }, +} + +data example_compute_fabric: ComputeFabric = ComputeFabric { + pool: example_heavy_compute_pool, + remote_executor: example_remote_executor, + local_offer: example_resource_offer, +} + +data example_remote_request: ComputeRequest = ComputeRequest { + requester: example_session_subject, + workload: ExecCommand { command: "cargo build --workspace" }, + needs: heavy_compute_lane_work_demand.resources, + locality: FulfillableRemotely, +} + +data example_local_request: ComputeRequest = ComputeRequest { + requester: example_session_subject, + workload: ExecCommand { command: "profile_typed_ast_byte_breakdown" }, + needs: heavy_compute_lane_work_demand.resources, + locality: MustRunLocal, +} + +data example_oversized_local_request: ComputeRequest = ComputeRequest { + requester: example_session_subject, + workload: ExecCommand { command: "oversized-local" }, + needs: example_thread_work_demand.resources, + locality: MustRunLocal, +} + +data example_gha_runner_request: ComputeRequest = ComputeRequest { + requester: example_session_subject, + workload: GhaRunner { repo: "gunb-ai/gunbc", labels: ["self-hosted", "srv1"] }, + needs: heavy_compute_lane_work_demand.resources, + locality: MustRunLocal, +} + +fn witness_session_subject_is_control_plane_agent() -> Bool { + match example_session_subject.agent { + ClaudeCode => true + Codex => true + Cursor => true + } +} + +fn witness_remote_request_fulfilled_off_session() -> Bool { + let outcome = fulfill( + fabric: example_compute_fabric, + ledger: empty_lease_ledger, + request: example_remote_request, + lease_id: "unused-remote" + ) + match outcome.fulfillment { + Fulfilled { executor: e } => + match e.target { + BuildBuddyRemote { endpoint: _ } => ledger_held_count(ledger: outcome.ledger) == 0 + LeasedComputeLane { lease_id: _, host: _ } => false + } + FulfillmentQueued { lane_count: _, held: _ } => false + FulfillmentRejected { reason: _ } => false + } +} + +fn witness_local_request_leases_a_lane() -> Bool { + let outcome = fulfill( + fabric: example_compute_fabric, + ledger: empty_lease_ledger, + request: example_local_request, + lease_id: "lease-local-1" + ) + match outcome.fulfillment { + Fulfilled { executor: e } => + match e.target { + LeasedComputeLane { lease_id: lid, host: _ } => + lid == "lease-local-1" && ledger_held_count(ledger: outcome.ledger) == 1 + BuildBuddyRemote { endpoint: _ } => false + } + FulfillmentQueued { lane_count: _, held: _ } => false + FulfillmentRejected { reason: _ } => false + } +} + +fn witness_local_request_rejected_when_pool_exhausted() -> Bool { + let o1 = fulfill( + fabric: example_compute_fabric, ledger: empty_lease_ledger, request: example_local_request, + lease_id: "l1" + ) + let o2 = fulfill( + fabric: example_compute_fabric, ledger: o1.ledger, request: example_local_request, + lease_id: "l2" + ) + let o3 = fulfill( + fabric: example_compute_fabric, ledger: o2.ledger, request: example_local_request, + lease_id: "l3" + ) + match o3.fulfillment { + FulfillmentRejected { reason: r } => + match r { + LocalCapacityExhausted { lane_count: _, held: h } => + h == 2 && ledger_held_count(ledger: o3.ledger) == 2 + NeedsUnsatisfiable { reason: _ } => false + DuplicateLeaseId { lease_id: _ } => false + } + Fulfilled { executor: _ } => false + FulfillmentQueued { lane_count: _, held: _ } => false + } +} + +fn witness_unsatisfiable_local_needs_rejected() -> Bool { + let outcome = fulfill( + fabric: example_compute_fabric, + ledger: empty_lease_ledger, + request: example_oversized_local_request, + lease_id: "oversized" + ) + match outcome.fulfillment { + FulfillmentRejected { reason: r } => + match r { + NeedsUnsatisfiable { reason: m } => m.dimension == DemandCpu + LocalCapacityExhausted { lane_count: _, held: _ } => false + DuplicateLeaseId { lease_id: _ } => false + } + Fulfilled { executor: _ } => false + FulfillmentQueued { lane_count: _, held: _ } => false + } +} + +fn witness_gha_runner_request_leases_same_lane_as_exec() -> Bool { + let exec_outcome = fulfill( + fabric: example_compute_fabric, ledger: empty_lease_ledger, request: example_local_request, + lease_id: "exec-1" + ) + let runner_outcome = fulfill( + fabric: example_compute_fabric, ledger: empty_lease_ledger, request: example_gha_runner_request, + lease_id: "runner-1" + ) + let same_envelope = work_demand_content_digest(demand: request_to_work_demand(request: example_local_request)) == work_demand_content_digest(demand: request_to_work_demand(request: example_gha_runner_request)) + match exec_outcome.fulfillment { + Fulfilled { executor: ee } => + match runner_outcome.fulfillment { + Fulfilled { executor: re } => + match ee.target { + LeasedComputeLane { lease_id: _, host: eh } => + match re.target { + LeasedComputeLane { lease_id: _, host: rh } => eh == rh && same_envelope + BuildBuddyRemote { endpoint: _ } => false + } + BuildBuddyRemote { endpoint: _ } => false + } + FulfillmentQueued { lane_count: _, held: _ } => false + FulfillmentRejected { reason: _ } => false + } + FulfillmentQueued { lane_count: _, held: _ } => false + FulfillmentRejected { reason: _ } => false + } +} + +fn witness_gha_runner_queues_when_pool_exhausted_not_rejects() -> Bool { + let o1 = fulfill( + fabric: example_compute_fabric, ledger: empty_lease_ledger, request: example_local_request, + lease_id: "l1" + ) + let o2 = fulfill( + fabric: example_compute_fabric, ledger: o1.ledger, request: example_local_request, + lease_id: "l2" + ) + let o3 = fulfill( + fabric: example_compute_fabric, ledger: o2.ledger, request: example_gha_runner_request, + lease_id: "runner-overflow" + ) + match o3.fulfillment { + FulfillmentQueued { lane_count: _, held: h } => + h == 2 && ledger_held_count(ledger: o3.ledger) == 2 + Fulfilled { executor: _ } => false + FulfillmentRejected { reason: _ } => false + } +} + +fn witness_gha_runner_uses_ci_lane_source() -> Bool { + match workload_lease_source(workload: GhaRunner { repo: "gunb-ai/gunbc", labels: [] }) { + GithubCiRunner => true + OnDemandSessionLease => false + } +} + +fn witness_heavy_pool_bytes_is_lanes_times_envelope() -> Bool { + byte_size_count(heavy_compute_pool_bytes(pool: example_heavy_compute_pool)) == 68719476736 +} + +fn witness_pool_unifies_ci_and_ondemand_then_rejects() -> Bool { + match lease_grant( + pool: example_heavy_compute_pool, + ledger: empty_lease_ledger, + offer: example_resource_offer, + lease_id: "ci-1", + source: GithubCiRunner, + demand: heavy_compute_lane_work_demand + ) { + LeaseRejectedIneligible { reason: _ } => false + LeaseRejectedPoolExhausted { lane_count: _, held: _ } => false + LeaseRejectedDuplicateId { lease_id: _ } => false + LeaseGranted { ledger: l1, granted: _ } => + match lease_grant( + pool: example_heavy_compute_pool, + ledger: l1, + offer: example_resource_offer, + lease_id: "od-1", + source: OnDemandSessionLease, + demand: heavy_compute_lane_work_demand + ) { + LeaseRejectedIneligible { reason: _ } => false + LeaseRejectedPoolExhausted { lane_count: _, held: _ } => false + LeaseRejectedDuplicateId { lease_id: _ } => false + LeaseGranted { ledger: l2, granted: _ } => + match lease_grant( + pool: example_heavy_compute_pool, + ledger: l2, + offer: example_resource_offer, + lease_id: "od-2", + source: OnDemandSessionLease, + demand: heavy_compute_lane_work_demand + ) { + LeaseGranted { ledger: _, granted: _ } => false + LeaseRejectedIneligible { reason: _ } => false + LeaseRejectedDuplicateId { lease_id: _ } => false + LeaseRejectedPoolExhausted { lane_count: _, held: h } => + ledger_held_count(ledger: l2) == 2 + && h == 2 + && ledger_holds_source(ledger: l2, source: GithubCiRunner) + && ledger_holds_source(ledger: l2, source: OnDemandSessionLease) + } + } + } +} + +fn witness_lease_release_frees_lane() -> Bool { + match lease_grant( + pool: example_heavy_compute_pool, + ledger: empty_lease_ledger, + offer: example_resource_offer, + lease_id: "rel-1", + source: GithubCiRunner, + demand: heavy_compute_lane_work_demand + ) { + LeaseRejectedIneligible { reason: _ } => false + LeaseRejectedPoolExhausted { lane_count: _, held: _ } => false + LeaseRejectedDuplicateId { lease_id: _ } => false + LeaseGranted { ledger: l1, granted: _ } => { + let l2 = lease_release(ledger: l1, lease_id: "rel-1") + ledger_held_count(ledger: l1) == 1 + && ledger_held_count(ledger: l2) == 0 + && ledger_has_free_lane(pool: example_heavy_compute_pool, ledger: l2) + } + } +} + +fn witness_full_ledger_conserves_with_no_free_lane() -> Bool { + match lease_grant( + pool: example_heavy_compute_pool, + ledger: empty_lease_ledger, + offer: example_resource_offer, + lease_id: "ci-1", + source: GithubCiRunner, + demand: heavy_compute_lane_work_demand + ) { + LeaseRejectedIneligible { reason: _ } => false + LeaseRejectedPoolExhausted { lane_count: _, held: _ } => false + LeaseRejectedDuplicateId { lease_id: _ } => false + LeaseGranted { ledger: l1, granted: _ } => + match lease_grant( + pool: example_heavy_compute_pool, + ledger: l1, + offer: example_resource_offer, + lease_id: "od-1", + source: OnDemandSessionLease, + demand: heavy_compute_lane_work_demand + ) { + LeaseRejectedIneligible { reason: _ } => false + LeaseRejectedPoolExhausted { lane_count: _, held: _ } => false + LeaseRejectedDuplicateId { lease_id: _ } => false + LeaseGranted { ledger: l2, granted: _ } => + ledger_conserves(pool: example_heavy_compute_pool, ledger: l2) + && !ledger_has_free_lane(pool: example_heavy_compute_pool, ledger: l2) + && ledger_held_count(ledger: l2) == 2 + } + } +} + +fn witness_duplicate_lease_id_rejected_fail_closed() -> Bool { + let o1 = fulfill( + fabric: example_compute_fabric, ledger: empty_lease_ledger, request: example_local_request, + lease_id: "dup-1" + ) + let o2 = fulfill( + fabric: example_compute_fabric, ledger: o1.ledger, request: example_local_request, + lease_id: "dup-1" + ) + match o2.fulfillment { + FulfillmentRejected { reason: r } => + match r { + DuplicateLeaseId { lease_id: lid } => + lid == "dup-1" && ledger_held_count(ledger: o2.ledger) == 1 + LocalCapacityExhausted { lane_count: _, held: _ } => false + NeedsUnsatisfiable { reason: _ } => false + } + Fulfilled { executor: _ } => false + FulfillmentQueued { lane_count: _, held: _ } => false + } +} diff --git a/dsl/test/claim/compute_fabric_resource_witness_test.dag b/dsl/test/claim/compute_fabric_resource_witness_test.dag index ca14963d86e..cee2fe6d04b 100644 --- a/dsl/test/claim/compute_fabric_resource_witness_test.dag +++ b/dsl/test/claim/compute_fabric_resource_witness_test.dag @@ -8,6 +8,19 @@ import product.compute_fabric { witness_high_demand_shrinks_envelope_and_fits, witness_fail_open_memory_blind_demand_exceeds_budget, witness_unreadable_budget_uses_conservative_demand, + witness_session_subject_is_control_plane_agent, + witness_remote_request_fulfilled_off_session, + witness_local_request_leases_a_lane, + witness_local_request_rejected_when_pool_exhausted, + witness_unsatisfiable_local_needs_rejected, + witness_duplicate_lease_id_rejected_fail_closed, + witness_gha_runner_request_leases_same_lane_as_exec, + witness_gha_runner_queues_when_pool_exhausted_not_rejects, + witness_gha_runner_uses_ci_lane_source, + witness_heavy_pool_bytes_is_lanes_times_envelope, + witness_pool_unifies_ci_and_ondemand_then_rejects, + witness_lease_release_frees_lane, + witness_full_ledger_conserves_with_no_free_lane, } test fn compute_fabric_resource_feasibility_holds() -> Bool { @@ -22,3 +35,25 @@ test fn ci_floor_run_demand_envelope_model_holds() -> Bool { && witness_fail_open_memory_blind_demand_exceeds_budget() && witness_unreadable_budget_uses_conservative_demand() } + +test fn compute_request_fulfillment_interface_holds() -> Bool { + witness_session_subject_is_control_plane_agent() + && witness_remote_request_fulfilled_off_session() + && witness_local_request_leases_a_lane() + && witness_local_request_rejected_when_pool_exhausted() + && witness_unsatisfiable_local_needs_rejected() + && witness_duplicate_lease_id_rejected_fail_closed() +} + +test fn gha_runner_is_same_request_on_one_ledger() -> Bool { + witness_gha_runner_request_leases_same_lane_as_exec() + && witness_gha_runner_queues_when_pool_exhausted_not_rejects() + && witness_gha_runner_uses_ci_lane_source() +} + +test fn fulfillment_internals_conserve_the_pool() -> Bool { + witness_heavy_pool_bytes_is_lanes_times_envelope() + && witness_pool_unifies_ci_and_ondemand_then_rejects() + && witness_lease_release_frees_lane() + && witness_full_ledger_conserves_with_no_free_lane() +}