Can We Build an Open-Source Palantir?

Can We Build an Open-Source Palantir?

This post introduces Open Intelligence Grid, an experimental open-source project exploring whether we can build a public-interest data platform inspired by Palantir-style intelligence systems and SETI-style distributed computing. The goal is to imagine, and begin building, a transparent civic tool that helps communities map public information, coordinate resources, and process open datasets through a network of volunteer computers.

Disclaimer: This project is an early alpha prototype, not a finished product, not a security-hardened platform, and not an official alternative to Palantir. It should only be used with public, synthetic, or safely sanitized data. Do not use it for private records, medical information, financial data, personal communications, law enforcement files, or anything that could expose or harm real people. The purpose is education, experimentation, and public-good technology, not surveillance

Can We Build an Open-Source Palantir?

I asked a dangerous little question:

Could we build an open-source version of Palantir?

Not a knockoff. Not a spy toy. Not some shadowy data dragon sleeping under a government contract. I mean a free, public, open-source intelligence platform that communities, researchers, nonprofits, journalists, local governments, disaster-response teams, educators, and ordinary curious humans could use without having to sell their souls to the Great Dashboard Goblin.

Then I asked the second question, which is where the wires started glowing:

Could the processing be offloaded to a network of computers, kind of like the old SETI project?

Could unused computers help process public data for the public good?

The answer is yes.

Not easy yes. Not weekend-project yes. Not “just add blockchain and sprinkle with venture-capital paprika” yes.

But real yes.

So I started building it.

The project is called:

Open Intelligence Grid

Open Intelligence Grid is an open-source starter platform for civic intelligence, distributed analysis, and volunteer computing.

The basic idea is simple:

Take the useful parts of a Palantir-like system, make them open, transparent, auditable, and community-owned, then combine them with a distributed worker network inspired by projects like SETI@home.

Instead of one giant private intelligence machine owned by a company, imagine a public-interest data platform that helps people understand the world, coordinate action, and process large public datasets without locking everything behind a billion-dollar contract.

That is the dream.

Version 0.1.0 is the first brick in that cathedral.

What Palantir-Like Really Means

When people say “an open-source Palantir,” they often imagine one giant magical database where every fact in the universe gets poured into a chrome cauldron and turned into predictive lightning.

That is not quite right.

A Palantir-like system is not just a database. It is a system for connecting information to real-world objects, relationships, actions, permissions, and decisions.

That means it needs to understand things like:

  • People
  • Organizations
  • Locations
  • Programs
  • Events
  • Resources
  • Transactions
  • Equipment
  • Documents
  • Relationships
  • Workflows
  • Permissions
  • Audit trails

The real power is not just storing information. The real power is turning scattered data into a usable model of reality.

For example, a community version could map:

  • Which nonprofits provide food assistance
  • Where those programs operate
  • What hours they are open
  • Who qualifies
  • What resources are available
  • Which areas are underserved
  • What public datasets support the analysis
  • What changed over time
  • Who updated the information
  • What actions were taken because of it

That is not evil by default. A hammer can build a shelter or crack a skull. The moral question is who holds the hammer, who gets hit, who gets housed, and whether anyone is allowed to inspect the toolbox.

Open Intelligence Grid is designed around transparency, public benefit, and shared ownership.

The SETI-Style Part

The old SETI@home model was beautifully weird.

Millions of ordinary computers helped analyze radio telescope data in the background while their owners slept, worked, or wandered into the kitchen for crackers.

That model is powerful because many big problems can be broken into smaller chunks.

A central coordinator hands out work. Volunteer computers process their assigned pieces. Results come back. The system checks whether the answers match.

That is not quite a peer-to-peer mesh. It is more like a distributed public workbench.

For Open Intelligence Grid, this is the safer first step.

Instead of trying to build a full decentralized mesh from day one, the first version uses a central coordinator and distributed workers.

That coordinator:

  • Creates jobs
  • Splits work into replicated units
  • Sends work to registered workers
  • Receives results
  • Hashes and compares outputs
  • Builds consensus
  • Handles disagreement
  • Records audit events

The worker computers do the processing.

This is how you get useful distributed computing without immediately releasing a cybernetic raccoon into the electrical grid.

What Version 0.1.0 Can Do

Open Intelligence Grid v0.1.0 is a working alpha release.

It includes:

  • A FastAPI coordinator
  • A basic ontology system
  • Entity types
  • Entities with flexible JSON properties
  • Typed relationships between entities
  • One-hop graph exploration
  • Worker registration
  • Distributed job creation
  • Replicated work units
  • Expiring leases
  • Result hashing
  • Consensus validation
  • Third-worker tie-breaking when results disagree
  • Worker reputation foundations
  • Audit event recording
  • SQLite support for local experimentation
  • PostgreSQL and Docker Compose support
  • A minimal browser dashboard
  • Command-line worker tools
  • Automated tests
  • MIT open-source licensing
  • Documentation for setup, architecture, security, governance, and contribution

This is not vaporware. This is source code, documentation, tests, and a working prototype.

It is small, but it breathes.

What the Worker Network Does

The worker client can currently process simple deterministic jobs.

Right now, it supports:

  • Word counting
  • Integer summation
  • Text hashing
  • JSON structure analysis

That may sound humble, but this is the seed crystal.

The important thing is not that it can count words. The important thing is that it can:

  1. Register a worker
  2. Receive work
  3. Process the work independently
  4. Submit a result
  5. Compare that result against another worker’s result
  6. Reach verified consensus
  7. Detect disagreement
  8. Create additional work to break a tie

That is the beginning of a real distributed system.

A worker network needs to distrust itself intelligently. Random computers on the internet cannot simply be believed. They may fail, lie, crash, cheat, or return nonsense because someone’s gaming PC overheated under a desk covered in dust bunnies and Mountain Dew fossils.

So the coordinator does not accept a single answer as truth.

It asks multiple workers.

If they match, the job is verified.

If they disagree, the system creates another work unit and waits for a majority.

That is the tiny parliament of machines.

Why This Should Not Process Private Data Yet

This part matters.

Volunteer computers should not process sensitive information.

Not medical records. Not private financial data. Not personal communications. Not criminal justice records. Not anything that could expose or harm vulnerable people.

Open Intelligence Grid v0.1.0 is meant for:

  • Public datasets
  • Synthetic data
  • Open research data
  • Sanitized documents
  • Community resource directories
  • Environmental data
  • Public records
  • Educational experiments
  • Non-sensitive civic analysis

The dream is not to build a surveillance machine with better branding.

The dream is to build a public-interest intelligence system that helps people understand complicated systems without surrendering power to private black boxes.

Sensitive data would require a different architecture, such as federated processing, where the computation goes to the organization that owns the data, and the raw data does not leave its home.

That comes later.

First, the system must be transparent, testable, and safe.

The First Architecture

The current architecture has four major parts.

1. The Coordinator

The coordinator is the brain of the system.

It manages:

  • Ontology definitions
  • Entities
  • Relationships
  • Jobs
  • Work units
  • Workers
  • Results
  • Consensus
  • Audit events

It exposes an API that other tools can use.

2. The Ontology Layer

The ontology layer lets the platform model real-world objects and their relationships.

For example:

  • An organization operates a program
  • A program is available at a location
  • A location serves a neighborhood
  • A resource has eligibility requirements
  • A public document supports a claim

This is what makes the system more than a pile of files.

The ontology is where raw data starts becoming usable knowledge.

3. The Worker Network

Workers are computers that volunteer to process work.

Each worker can:

  • Register with the coordinator
  • Request available work
  • Process the work locally
  • Submit a result
  • Build reputation over time

In the future, workers could process public documents, satellite image tiles, open data transformations, embeddings, simulations, deduplication tasks, and other jobs that can be safely split apart.

4. The Validation System

The validation system checks whether independent workers produce the same result.

If they do, the job can be marked complete.

If they do not, the coordinator asks another worker.

That is the first step toward trustworthy public computing.

What This Could Become

Open Intelligence Grid could eventually become useful for many public-interest projects.

Imagine using it to help map:

  • Food assistance networks
  • Homelessness services
  • Climate risks
  • Public health resources
  • Environmental pollution
  • Local infrastructure
  • Disaster response assets
  • Educational programs
  • Open scientific datasets
  • Municipal service gaps
  • Public grant opportunities
  • Mutual aid networks

A town, school, nonprofit, or citizen research group could run its own node.

Public data could be shared.

Private data could stay private.

Volunteer computers could help process open workloads.

Researchers and community groups could build tools on top of the shared platform.

Instead of intelligence being something done to people, intelligence could become something communities do for themselves.

That is the ethical turn.

That is the hinge.

What It Is Not Yet

Version 0.1.0 is not a finished Palantir replacement.

It does not yet include:

  • True peer-to-peer mesh networking
  • WebAssembly sandboxing
  • Signed job packages
  • Federated private-data queries
  • Organization-level permissions
  • Visual pipeline builders
  • Advanced geospatial tools
  • Live streaming data
  • AI model execution
  • Full production identity management
  • Database migrations
  • Professional security auditing
  • Byzantine fault tolerance
  • A polished user interface

This is the first working release, not the final beast.

It is a foundation.

The next major step is to make worker execution safer with signed, content-addressed task bundles and sandboxed runtimes. That would let workers process public jobs more securely, with fewer risks to the person donating compute and fewer risks to the system receiving results.

After that comes better federation, richer ontology tools, stronger permissions, better dashboards, and eventually more decentralized coordination.

Why Open Source Matters

If a system helps people make decisions, people should be able to inspect how it works.

If a system processes public data, the public should be able to understand the process.

If a system affects communities, communities should not be locked out of the machine room.

Open source does not magically make software ethical. A villain can still use a transparent hammer.

But open source gives us the possibility of inspection, repair, remixing, accountability, education, and public ownership.

That matters.

Especially now.

We live in an age where data is power, dashboards are priesthoods, and too many decisions arrive from sealed boxes with corporate logos stamped on the lid.

Open Intelligence Grid is a tiny rebellion against that.

A civic data machine should not have to be a dragon hoard.

It can be a workshop.

It can be a library.

It can be a barn raising with APIs.

The Bigger Dream

The long-term vision is a free intelligence platform for public good.

A system where:

  • Communities can map their own needs
  • Researchers can share open workloads
  • Nonprofits can coordinate services
  • Local governments can understand gaps
  • Volunteers can donate computing power
  • Sensitive data can remain protected
  • Every transformation can be audited
  • Every major decision can be traced
  • Every user can inspect the code

Not a surveillance empire.

Not a black box.

Not a techno-oracle wearing a badge.

A transparent intelligence grid for people who want to build, help, understand, and organize.

That is the dream.

Version 0.1.0 is only the first little campfire.

But every signal fire starts somewhere.

Project Status

Open Intelligence Grid v0.1.0 has working source code, tests, documentation, and a release package.

The project includes:

  • Source code
  • API documentation
  • Quickstart guide
  • Architecture guide
  • Security notes
  • Threat model
  • Governance guide
  • Contribution guide
  • Code of conduct
  • Roadmap
  • MIT license

The plan is to upload the code for others to use, study, modify, and improve for free.

GitHub link:

[INSERT GITHUB LINK HERE]

Final Thought

The question was simple:

Can we build an open-source version of Palantir with distributed computing?

The answer is yes.

But the better question is:

Can we build a public intelligence system that refuses to become a monster?

That answer is not just technical.

That answer depends on architecture, governance, ethics, transparency, and who we invite to help build it.

So here is the invitation:

Let’s build the machine in public.

Let’s keep the code open.

Let’s keep the data boundaries clear.

Let’s make the workers useful, the coordinator accountable, the ontology humane, and the whole strange contraption available to anyone who wants to use intelligence for repair instead of domination.

The grid is open.

The little machines are waking up.

Let’s teach them to help.

 

Now to the code. This is of course rough. Ment as a starting point. 

The best one for your blog is the Markdown source-code appendix. It has every file labeled by path with code blocks.

The best one for developers is the shell script. Someone can paste/run it and it will recreate the entire repository structure, files, tests, docs, and config.

Shell script:

#!/usr/bin/env bash
set -euo pipefail

PROJECT_DIR="${1:-open-intelligence-grid}"
mkdir -p "$PROJECT_DIR"
cd "$PROJECT_DIR"

echo "Creating Open Intelligence Grid v0.1.0 source tree in: $(pwd)"

cat > '.dockerignore' <<'OIG_FILE_001_05a721eb'
.git
.venv
__pycache__
.pytest_cache
*.db
.env
OIG_FILE_001_05a721eb

cat > '.env.example' <<'OIG_FILE_002_d4dae00d'
OIG_APP_NAME=Open Intelligence Grid
OIG_ENVIRONMENT=development
OIG_DATABASE_URL=sqlite:///./oig.db
OIG_ADMIN_KEY=change-me-now
OIG_LEASE_SECONDS=300
OIG_ALLOW_SAME_WORKER_REPLICAS=false
OIG_CORS_ORIGINS=[]
OIG_FILE_002_d4dae00d

mkdir -p '.github/workflows'
cat > '.github/workflows/ci.yml' <<'OIG_FILE_003_899ce9c2'
name: CI

on:
  push:
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.11", "3.12", "3.13"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
      - run: python -m pip install --upgrade pip
      - run: pip install -e '.[dev]'
      - run: ruff check src tests
      - run: pytest
OIG_FILE_003_899ce9c2

cat > '.gitignore' <<'OIG_FILE_004_a5cc2925'
.venv/
__pycache__/
*.py[cod]
.pytest_cache/
.ruff_cache/
.coverage
htmlcov/
.env
*.db
*.sqlite3
dist/
build/
*.egg-info/
.DS_Store
OIG_FILE_004_a5cc2925

cat > 'CHANGELOG.md' <<'OIG_FILE_005_ab09011f'
# Changelog

## 0.1.0 - 2026-07-01

### Added

- Ontology object types, entities, and relationships.
- One-hop graph traversal.
- Worker registration and bearer-token authentication.
- Database-backed work leasing.
- Replicated deterministic jobs with hash-based consensus.
- Tie-breaking work units for conflicting results.
- SQLite and PostgreSQL support.
- Docker Compose environment.
- Minimal browser dashboard.
- Tests and initial security documentation.
OIG_FILE_005_ab09011f

cat > 'CODE_OF_CONDUCT.md' <<'OIG_FILE_006_0834ae01'
# Community code of conduct

## Our standard

Participants should help create a technically serious, curious, and humane community. Expected behavior includes:

- Critiquing ideas and code without degrading people.
- Welcoming contributors with different levels of experience.
- Documenting decisions that affect users or communities.
- Respecting privacy, consent, and the right to correct errors.
- Disclosing conflicts of interest.
- Avoiding harassment, threats, discrimination, and deliberate disruption.

## Public-interest responsibility

This project can become powerful. Contributors must not use project spaces to promote covert surveillance, harassment, discriminatory targeting, or collection of personal information without a lawful and ethical basis.

## Enforcement

Before public launch, maintainers should publish a dedicated conduct contact, a private reporting method, an appeal procedure, and a transparent range of enforcement actions.
OIG_FILE_006_0834ae01

cat > 'CONTRIBUTING.md' <<'OIG_FILE_007_3f454a98'
# Contributing

Thank you for helping build an inspectable public-interest data platform.

## Development setup

```bash
python -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]'
pytest
ruff check src tests
```

## Contribution rules

1. Keep pull requests focused and documented.
2. Add tests for behavior changes.
3. Do not add surveillance-specific features without a public-interest review.
4. Do not add dependencies without explaining why the standard library or an existing dependency is insufficient.
5. Never commit secrets or real personal data.
6. Preserve backward compatibility or document a migration path.

## Commit suggestions

Use clear prefixes such as `feat:`, `fix:`, `docs:`, `test:`, and `security:`.

## Definition of done

A change should include implementation, tests, documentation, and a note in `CHANGELOG.md` when users will notice it.
OIG_FILE_007_3f454a98

cat > 'Dockerfile' <<'OIG_FILE_008_6651ddff'
FROM python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

WORKDIR /app

COPY pyproject.toml README.md LICENSE ./
COPY src ./src
RUN pip install --no-cache-dir .

EXPOSE 8000
CMD ["uvicorn", "oig.main:app", "--host", "0.0.0.0", "--port", "8000"]
OIG_FILE_008_6651ddff

cat > 'GOVERNANCE.md' <<'OIG_FILE_009_2633952a'
# Governance draft

OIG should eventually be governed by a nonprofit or multistakeholder foundation rather than a single vendor.

## Proposed bodies

- **Maintainers:** accept code and manage releases.
- **Security team:** coordinates private vulnerability reports.
- **Public-interest council:** reviews high-risk capabilities and deployment patterns.
- **Data-governance working group:** develops consent, retention, provenance, and redress standards.
- **Community assembly:** proposes roadmap items and elects or confirms council members.

## High-risk change review

Features involving identity resolution, location tracking, biometric inference, predictive policing, mass scraping, or covert monitoring should require an explicit written review and documented safeguards before merging.

This is a starting constitution, not stone tablets descending from a venture-capital cloud.
OIG_FILE_009_2633952a

cat > 'LICENSE' <<'OIG_FILE_010_0398ccd0'
MIT License

Copyright (c) 2026 Open Intelligence Grid contributors

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
OIG_FILE_010_0398ccd0

cat > 'Makefile' <<'OIG_FILE_011_836efb6e'
.PHONY: install dev test lint run docker-up docker-down

install:
python -m pip install -e .

dev:
python -m pip install -e '.[dev]'

test:
pytest

lint:
ruff check src tests

run:
oig-server

docker-up:
docker compose up --build

docker-down:
docker compose down
OIG_FILE_011_836efb6e

cat > 'README.md' <<'OIG_FILE_012_8ec9a00b'
# Open Intelligence Grid

**Open Intelligence Grid (OIG)** is an early, working open-source foundation for a transparent data-and-operations platform with distributed volunteer computing.

It combines two ideas:

1. An **ontology layer** that represents real-world object types, entities, and relationships.
2. A **coordinator/worker network** that divides deterministic jobs into independently validated work units.

This is a usable alpha, not a production replacement for Palantir Foundry. It is intentionally small enough to understand, run, audit, and improve.

## What works in v0.1.0

- Define ontology object types with JSON Schema-like property descriptions.
- Create entities and relationships.
- Traverse a one-hop entity graph.
- Register authenticated volunteer workers.
- Submit distributed jobs with configurable replication.
- Lease work units to separate workers.
- Validate results by canonical-hash consensus.
- Add a tie-breaking replica when workers disagree.
- Record basic audit events.
- Run on SQLite for development or PostgreSQL with Docker Compose.
- Use the bundled deterministic worker tasks:
  - `word_count`
  - `sum_numbers` (integer inputs)
  - `hash_text`
  - `json_stats`

## Safety boundary

Do **not** put secrets, personal records, medical records, private messages, credentials, or other sensitive data into volunteer-computing jobs. Workers must be treated as untrusted machines. OIG v0.1.0 is for public, synthetic, sanitized, or otherwise safe-to-distribute inputs.

Read [SECURITY.md](SECURITY.md) and [docs/THREAT_MODEL.md](docs/THREAT_MODEL.md) before exposing a deployment to the internet.

## Quick start with Python

Requirements: Python 3.11 or newer.

```bash
python -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]'
cp .env.example .env
oig-server
```

Open:

- API documentation: `http://localhost:8000/docs`
- Minimal dashboard: `http://localhost:8000/`
- Health check: `http://localhost:8000/health`

The development admin key in `.env.example` is `change-me-now`. Change it before doing anything beyond local testing.

## Quick start with Docker

```bash
cp .env.example .env
docker compose up --build
```

The Compose setup starts PostgreSQL and the API.

## Run the demo

With the server running:

```bash
python examples/bootstrap_demo.py
```

The script creates an ontology, two workers, and a replicated word-count job. It prints two worker tokens. In two terminals, run:

```bash
oig-worker --server http://localhost:8000 --token WORKER_ONE_TOKEN --once
oig-worker --server http://localhost:8000 --token WORKER_TWO_TOKEN --once
```

Then inspect the job:

```bash
curl http://localhost:8000/v1/jobs/JOB_ID \
  -H 'X-Admin-Key: change-me-now'
```

## Repository map

```text
src/oig/
  api.py          HTTP routes
  config.py       Environment-based settings
  db.py           SQLAlchemy engine/session setup
  models.py       Persistent ontology, jobs, workers, and audit records
  schemas.py      Request and response models
  security.py     Admin and worker authentication
  services.py     Leasing, consensus, and graph services
  tasks.py        Safe deterministic worker task implementations
  worker.py       Volunteer worker client
  main.py         Application entry point and dashboard

tests/            API and consensus tests
docs/             Architecture, API, threat model, and roadmap
examples/         Runnable bootstrap example
```

## Guiding principles

- Public-interest uses before surveillance uses.
- Data minimization before data accumulation.
- Traceable transformations before black boxes.
- Compute moves toward sensitive data; sensitive data does not move toward strangers.
- No cryptocurrency is required for participation.
- Human governance is part of the architecture, not an afterthought.

## Current limitations

- The scheduler uses a relational database, not a peer-to-peer mesh.
- API-key authentication is suitable only for development and controlled pilots.
- There is no row-level policy engine yet.
- There are no signed job bundles or WebAssembly sandboxes yet.
- The ontology schema is stored and documented but not fully JSON-Schema validated.
- Database migrations are not yet included.
- Work leasing is deliberately simple and needs stronger transaction isolation at scale.

These limits are documented rather than hidden behind a cloud of enterprise incense.

## License

MIT. See [LICENSE](LICENSE).

## Working title

“Open Intelligence Grid” is a descriptive working title. Perform a proper trademark and project-name review before a major public launch.
OIG_FILE_012_8ec9a00b

cat > 'SECURITY.md' <<'OIG_FILE_013_7c875fef'
# Security policy

## Status

OIG v0.1.0 is an alpha research and development release. It has not received a professional security audit and must not be used as-is for classified, regulated, medical, financial, or personally identifying information.

## Non-negotiable data rule

Volunteer workers are untrusted. Never distribute secrets or sensitive source records to them. Encryption in transit does not change this rule because a worker must ordinarily access a task's plaintext input to calculate its output.

## Supported deployment posture

For the alpha release:

- Keep the API behind a trusted reverse proxy.
- Use TLS.
- Replace the development admin key.
- Restrict administrative endpoints by network and identity.
- Use PostgreSQL for shared deployments.
- Log and monitor worker registrations and job submissions.
- Use only public, synthetic, or sanitized job inputs.
- Run trusted data processing on organization-controlled workers.

## Reporting vulnerabilities

Do not open a public issue containing exploit details or exposed data. Before public launch, create a dedicated security contact and private disclosure channel in the repository's `SECURITY.md`.

## Planned controls

See [docs/ROADMAP.md](docs/ROADMAP.md) for signed job bundles, scoped identities, policy enforcement, sandboxed execution, provenance signatures, and federation.
OIG_FILE_013_7c875fef

cat > 'compose.yaml' <<'OIG_FILE_014_2701071a'
services:
  db:
    image: postgres:17-alpine
    environment:
      POSTGRES_DB: oig
      POSTGRES_USER: oig
      POSTGRES_PASSWORD: oig-development-password
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U oig -d oig"]
      interval: 5s
      timeout: 5s
      retries: 10
    volumes:
      - oig-postgres:/var/lib/postgresql/data

  api:
    build: .
    depends_on:
      db:
        condition: service_healthy
    environment:
      OIG_ENVIRONMENT: development
      OIG_DATABASE_URL: postgresql+psycopg://oig:oig-development-password@db:5432/oig
      OIG_ADMIN_KEY: ${OIG_ADMIN_KEY:-change-me-now}
      OIG_LEASE_SECONDS: ${OIG_LEASE_SECONDS:-300}
      OIG_ALLOW_SAME_WORKER_REPLICAS: ${OIG_ALLOW_SAME_WORKER_REPLICAS:-false}
    ports:
      - "8000:8000"

volumes:
  oig-postgres:
OIG_FILE_014_2701071a

mkdir -p 'docs'
cat > 'docs/API.md' <<'OIG_FILE_015_f5dea8f5'
# API overview

Interactive OpenAPI documentation is available at `/docs` while the server is running.

## Authentication

Administrative endpoints require:

```http
X-Admin-Key: your-admin-key
```

Worker endpoints require:

```http
Authorization: Bearer worker-token
```

## Important endpoints

| Method | Path | Purpose |
|---|---|---|
| GET | `/health` | Liveness check |
| GET | `/v1/stats` | Counts and job status summary |
| POST | `/v1/object-types` | Create an ontology object type |
| POST | `/v1/entities` | Create an entity |
| POST | `/v1/relationships` | Create a relationship |
| GET | `/v1/entities/{id}/graph` | Read a one-hop graph |
| POST | `/v1/workers/register` | Register a worker and receive its token |
| POST | `/v1/jobs` | Create a replicated job |
| POST | `/v1/work/claim` | Lease work to an authenticated worker |
| POST | `/v1/work/{id}/submit` | Submit a completed result |
| POST | `/v1/work/{id}/fail` | Report a failed work unit |

## Create a job

```bash
curl -X POST http://localhost:8000/v1/jobs \
  -H 'Content-Type: application/json' \
  -H 'X-Admin-Key: change-me-now' \
  -d '{
    "task_type": "sum_numbers",
    "input_payload": {"numbers": [1, 2, 3, 4]},
    "replicas_required": 2,
    "max_replicas": 3
  }'
```
OIG_FILE_015_f5dea8f5

mkdir -p 'docs'
cat > 'docs/ARCHITECTURE.md' <<'OIG_FILE_016_fd45feca'
# Architecture

## Current shape

```text
                         +-----------------------+
                         |   Browser / API user  |
                         +-----------+-----------+
                                     |
                                     v
+----------------+        +----------+-----------+        +----------------+
| Volunteer      | <----> | OIG Coordinator API | <----> | PostgreSQL or  |
| Worker A       |  HTTPS | FastAPI + services  |  SQL   | SQLite         |
+----------------+        +----------+-----------+        +----------------+
                                     ^
+----------------+                   |
| Volunteer      | <-----------------+
| Worker B       |
+----------------+
```

The coordinator is centralized in v0.1.0. Workers poll for jobs, lease a work unit, execute a known deterministic task, and submit a result. The coordinator accepts a job when enough independent workers produce the same canonical result hash.

## Ontology model

- `ObjectType` describes a class of real-world object and its expected properties.
- `Entity` is an instance of an object type.
- `Relationship` is a directed, typed edge between two entities.

The alpha stores property descriptions and entity properties as JSON. A later release should validate values against JSON Schema and add versioned ontology migrations.

## Job model

- `Job` contains a task type, input payload, validation threshold, and result.
- `WorkUnit` is one independently executed replica of a job.
- `Worker` has a bearer token, capabilities, activity state, and simple trust score.

A worker ordinarily cannot receive two replicas of the same job. Development deployments may override this with `OIG_ALLOW_SAME_WORKER_REPLICAS=true`.

## Consensus

Results are serialized as canonical JSON and hashed with SHA-256. A job completes when one hash has at least `replicas_required` completed replicas. If all issued replicas finish without consensus and the job has not reached `max_replicas`, the coordinator adds a tie-breaker work unit. Otherwise, the job becomes `disputed`.

This detects accidental and some malicious errors. It does not prove that a majority of workers are honest.

## Evolution path

1. Relational coordinator with trusted HTTPS workers.
2. Signed task manifests and sandboxed WebAssembly execution.
3. Organization-operated trusted worker pools.
4. Federated query and model execution near private data.
5. Replicated coordinators and append-only provenance logs.
6. Optional peer discovery and content-addressed task distribution.

Mesh networking belongs near the end because decentralizing an unclear protocol merely distributes the confusion.
OIG_FILE_016_fd45feca

mkdir -p 'docs'
cat > 'docs/GITHUB_PUBLISHING.md' <<'OIG_FILE_017_50182df8'
# Publishing this repository on GitHub

## 1. Review the working title

“Open Intelligence Grid” is a working title. Search existing projects and trademarks before promoting it as a permanent brand.

## 2. Create an empty repository

Create a new repository without automatically adding a README, license, or `.gitignore`, because those files already exist here.

## 3. Initialize and push

```bash
git init
git add .
git commit -m "Initial open-source alpha"
git branch -M main
git remote add origin YOUR_REPOSITORY_URL
git push -u origin main
```

## 4. Configure repository protections

Recommended settings:

- Require pull requests before merging to `main`.
- Require the CI workflow to pass.
- Require at least one approving review.
- Block force pushes and branch deletion.
- Enable secret scanning and dependency alerts where available.
- Create a private security-reporting channel before inviting broad use.

## 5. Create the first release

Tag the tested alpha:

```bash
git tag -a v0.1.0 -m "Open Intelligence Grid alpha"
git push origin v0.1.0
```

Attach the source archive and describe the current safety boundaries prominently. Do not present the alpha as suitable for sensitive or regulated data.

## 6. Suggested repository description

> Open-source ontology, data coordination, and independently validated volunteer-computing foundation.

## 7. Suggested initial topics

`open-source`, `distributed-computing`, `volunteer-computing`, `ontology`, `fastapi`, `postgresql`, `civic-tech`, `data-governance`
OIG_FILE_017_50182df8

mkdir -p 'docs'
cat > 'docs/QUICKSTART.md' <<'OIG_FILE_018_ef67bedb'
# Quickstart

## 1. Install

```bash
python -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]'
cp .env.example .env
```

## 2. Start the coordinator

```bash
oig-server
```

## 3. Bootstrap a demonstration

```bash
python examples/bootstrap_demo.py
```

Copy the two printed worker tokens and job ID.

## 4. Run two independent workers

Terminal A:

```bash
oig-worker --server http://localhost:8000 --token TOKEN_A --once
```

Terminal B:

```bash
oig-worker --server http://localhost:8000 --token TOKEN_B --once
```

## 5. Read the accepted job

```bash
curl http://localhost:8000/v1/jobs/JOB_ID \
  -H 'X-Admin-Key: change-me-now'
```

The job should have status `completed` and two matching completed work units.
OIG_FILE_018_ef67bedb

mkdir -p 'docs'
cat > 'docs/RELEASE_VALIDATION.md' <<'OIG_FILE_019_dc04a500'
# Release validation

Release: `0.1.0`

Validation performed on July 1, 2026 with Python 3.13.

## Automated checks

```text
pytest: 4 passed
ruff: all checks passed
compileall: passed
compose.yaml parse: passed
editable package installation: passed
```

## End-to-end check

A local coordinator was started with SQLite. The demonstration script then:

1. Created an ontology object type and entity.
2. Registered two distinct workers.
3. Created a word-count job requiring two matching replicas.
4. Ran each worker once through the command-line client.
5. Received two identical canonical result hashes.
6. Marked the job `completed` with the verified result.

## Important scope statement

These checks demonstrate that the alpha functions as designed in a local development environment. They are not a security audit, load test, formal protocol verification, or production-readiness certification.
OIG_FILE_019_dc04a500

mkdir -p 'docs'
cat > 'docs/ROADMAP.md' <<'OIG_FILE_020_3ea46177'
# Roadmap

## Phase 1: Harden the alpha

- Alembic database migrations.
- Pagination and filtering.
- Idempotency keys.
- Rate limiting.
- Structured logs, metrics, and traces.
- Stronger transaction-safe work claiming.
- JSON Schema validation for ontology entities.
- Capability and resource matching.
- Worker heartbeat and health screens.

## Phase 2: Safe distributed execution

- Signed task manifests.
- Content-addressed input bundles.
- WebAssembly/WASI task sandbox.
- CPU, memory, disk, and time quotas.
- Network-disabled tasks by default.
- Reproducible task builds.
- Random known-answer challenge jobs.
- Weighted consensus and worker reputation.

## Phase 3: Institutional federation

- Organizations, projects, and scoped roles.
- OpenID Connect authentication.
- Open Policy Agent integration.
- Data residency policies.
- Trusted worker pools.
- Federated aggregate queries.
- Differential privacy options.
- Signed lineage and provenance export.

## Phase 4: Operational applications

- Ontology actions and approval workflows.
- Event streams and alerting.
- Geospatial object support.
- Notebook and SQL workspaces.
- Visual pipeline builder.
- Dashboard/application SDK.
- Connectors for public datasets and common databases.

## Phase 5: Resilient network

- Replicated coordinators.
- Append-only transparent audit log.
- Peer discovery.
- Content-addressed peer-to-peer distribution for public task bundles.
- Offline-capable edge nodes.
- Byzantine-fault research and formal protocol specification.

A full Palantir-class platform is a long-range ecosystem. Each phase must remain independently useful rather than serving as scaffolding for a castle that never opens its doors.
OIG_FILE_020_3ea46177

mkdir -p 'docs'
cat > 'docs/THREAT_MODEL.md' <<'OIG_FILE_021_d3630528'
# Threat model

## Assets

- Ontology definitions and entity data.
- Administrative credentials.
- Worker identities and tokens.
- Job inputs and accepted results.
- Audit history and provenance.
- Coordinator availability.

## Adversaries

- A malicious volunteer worker returning fabricated results.
- Several colluding workers.
- An attacker stealing an administrator key.
- A coordinator operator abusing access.
- A user submitting a harmful or privacy-invasive job.
- A denial-of-service attacker creating excessive jobs or leases.

## Existing alpha mitigations

- Worker tokens are stored as SHA-256 hashes, not plaintext.
- Work leases expire.
- Separate worker identities are required for replicated results by default.
- Accepted results require matching canonical hashes.
- Administrative writes require an admin key.
- Audit events record major writes and worker actions.

## Known gaps

- No fine-grained roles or organization boundaries.
- No rate limiting.
- No cryptographic signature on task definitions.
- No secure execution sandbox.
- No remote attestation.
- No Sybil resistance beyond controlled worker registration.
- No immutable or externally witnessed audit log.
- No privacy-budget accounting or differential privacy.
- No content safety review for submitted workloads.

## Recommended pilot rules

1. Permit only an allowlist of task implementations.
2. Permit only public or synthetic inputs.
3. Manually approve workers.
4. Place the coordinator behind TLS and an identity-aware proxy.
5. Monitor task volume and unusual result disagreements.
6. Keep a human appeal and correction process for operational uses.
OIG_FILE_021_d3630528

mkdir -p 'examples'
cat > 'examples/bootstrap_demo.py' <<'OIG_FILE_022_afbd050f'
import os

import httpx

SERVER = os.getenv("OIG_SERVER", "http://localhost:8000").rstrip("/")
ADMIN_KEY = os.getenv("OIG_ADMIN_KEY", "change-me-now")
HEADERS = {"X-Admin-Key": ADMIN_KEY}


def post(path: str, payload: dict) -> dict:
    response = httpx.post(f"{SERVER}{path}", headers=HEADERS, json=payload, timeout=30)
    response.raise_for_status()
    return response.json()


def main() -> None:
    organization = post(
        "/v1/object-types",
        {
            "name": "Organization",
            "description": "A community organization or institution.",
            "schema_json": {
                "type": "object",
                "properties": {"name": {"type": "string"}, "city": {"type": "string"}},
                "required": ["name"],
            },
        },
    )
    entity = post(
        "/v1/entities",
        {
            "object_type_id": organization["id"],
            "external_id": "demo-library",
            "properties": {"name": "Demo Public Library", "city": "Gloucester"},
        },
    )

    workers = []
    for name in ("demo-worker-a", "demo-worker-b"):
        workers.append(
            post(
                "/v1/workers/register",
                {"name": name, "capabilities": ["word_count", "sum_numbers", "hash_text", "json_stats"]},
            )
        )

    job = post(
        "/v1/jobs",
        {
            "task_type": "word_count",
            "input_payload": {
                "text": "Public knowledge should glow brighter when more hands carry the lantern."
            },
            "replicas_required": 2,
            "max_replicas": 3,
        },
    )

    print("Created entity:", entity["id"])
    print("Created job:", job["id"])
    for worker in workers:
        print(f"{worker['name']} token: {worker['token']}")


if __name__ == "__main__":
    main()
OIG_FILE_022_afbd050f

cat > 'pyproject.toml' <<'OIG_FILE_023_5d07e7d7'
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "open-intelligence-grid"
version = "0.1.0"
description = "An open-source ontology and volunteer-computing coordination platform."
readme = "README.md"
requires-python = ">=3.11"
license = "MIT"
authors = [{ name = "Open Intelligence Grid contributors" }]
dependencies = [
  "fastapi>=0.115,<1.0",
  "uvicorn[standard]>=0.30,<1.0",
  "sqlalchemy>=2.0,<2.2",
  "pydantic>=2.8,<3.0",
  "pydantic-settings>=2.4,<3.0",
  "httpx>=0.27,<1.0",
  "psycopg[binary]>=3.2,<4.0",
]

[project.optional-dependencies]
dev = [
  "pytest>=8.0,<9.0",
  "pytest-cov>=5.0,<7.0",
  "ruff>=0.8,<1.0",
]

[project.scripts]
oig-server = "oig.main:run"
oig-worker = "oig.worker:main"

[tool.setuptools]
package-dir = {"" = "src"}

[tool.setuptools.packages.find]
where = ["src"]

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-q"

[tool.ruff]
line-length = 100
target-version = "py311"
OIG_FILE_023_5d07e7d7

mkdir -p 'src/oig'
cat > 'src/oig/__init__.py' <<'OIG_FILE_024_c7565b84'
"""Open Intelligence Grid core package."""

__version__ = "0.1.0"
OIG_FILE_024_c7565b84

mkdir -p 'src/oig'
cat > 'src/oig/api.py' <<'OIG_FILE_025_ce65a5ba'
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, selectinload

from .db import get_db
from .models import AuditEvent, Entity, Job, ObjectType, Relationship, Worker, WorkUnit
from .schemas import (
    AuditRead,
    EntityCreate,
    EntityRead,
    GraphRead,
    JobCreate,
    JobRead,
    ObjectTypeCreate,
    ObjectTypeRead,
    RelationshipCreate,
    RelationshipRead,
    WorkerRead,
    WorkerRegister,
    WorkerRegistered,
    WorkClaim,
    WorkFailure,
    WorkSubmit,
)
from .security import hash_token, issue_token, require_admin, require_worker
from .services import audit, create_job, fail_work, graph_for_entity, lease_work, submit_result
from .tasks import TASKS

router = APIRouter(prefix="/v1")
Admin = Annotated[str, Depends(require_admin)]
DB = Annotated[Session, Depends(get_db)]
AuthenticatedWorker = Annotated[Worker, Depends(require_worker)]


@router.get("/stats")
def stats(db: DB) -> dict:
    job_rows = db.execute(select(Job.status, func.count(Job.id)).group_by(Job.status)).all()
    return {
        "object_types": db.scalar(select(func.count(ObjectType.id))) or 0,
        "entities": db.scalar(select(func.count(Entity.id))) or 0,
        "relationships": db.scalar(select(func.count(Relationship.id))) or 0,
        "workers": db.scalar(select(func.count(Worker.id))) or 0,
        "jobs": dict(job_rows),
        "supported_tasks": sorted(TASKS),
    }


@router.post("/object-types", response_model=ObjectTypeRead, status_code=status.HTTP_201_CREATED)
def create_object_type(payload: ObjectTypeCreate, db: DB, _admin: Admin) -> ObjectType:
    item = ObjectType(**payload.model_dump(by_alias=True))
    db.add(item)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="object_type.created",
        target_type="object_type",
        target_id=item.id,
        details={"name": item.name},
    )
    try:
        db.commit()
    except IntegrityError as exc:
        db.rollback()
        raise HTTPException(status_code=409, detail="Object type name already exists") from exc
    db.refresh(item)
    return item


@router.get("/object-types", response_model=list[ObjectTypeRead])
def list_object_types(db: DB, limit: int = Query(default=100, ge=1, le=500)) -> list[ObjectType]:
    return list(db.scalars(select(ObjectType).order_by(ObjectType.name).limit(limit)).all())


@router.post("/entities", response_model=EntityRead, status_code=status.HTTP_201_CREATED)
def create_entity(payload: EntityCreate, db: DB, _admin: Admin) -> Entity:
    if db.get(ObjectType, payload.object_type_id) is None:
        raise HTTPException(status_code=404, detail="Object type not found")
    entity = Entity(**payload.model_dump())
    db.add(entity)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="entity.created",
        target_type="entity",
        target_id=entity.id,
        details={"object_type_id": entity.object_type_id},
    )
    try:
        db.commit()
    except IntegrityError as exc:
        db.rollback()
        raise HTTPException(status_code=409, detail="Duplicate external ID for object type") from exc
    db.refresh(entity)
    return entity


@router.get("/entities", response_model=list[EntityRead])
def list_entities(
    db: DB,
    object_type_id: str | None = None,
    limit: int = Query(default=100, ge=1, le=500),
) -> list[Entity]:
    statement = select(Entity).order_by(Entity.created_at.desc()).limit(limit)
    if object_type_id:
        statement = statement.where(Entity.object_type_id == object_type_id)
    return list(db.scalars(statement).all())


@router.get("/entities/{entity_id}", response_model=EntityRead)
def get_entity(entity_id: str, db: DB) -> Entity:
    entity = db.get(Entity, entity_id)
    if entity is None:
        raise HTTPException(status_code=404, detail="Entity not found")
    return entity


@router.post(
    "/relationships", response_model=RelationshipRead, status_code=status.HTTP_201_CREATED
)
def create_relationship(payload: RelationshipCreate, db: DB, _admin: Admin) -> Relationship:
    if db.get(Entity, payload.source_entity_id) is None:
        raise HTTPException(status_code=404, detail="Source entity not found")
    if db.get(Entity, payload.target_entity_id) is None:
        raise HTTPException(status_code=404, detail="Target entity not found")
    relationship = Relationship(**payload.model_dump())
    db.add(relationship)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="relationship.created",
        target_type="relationship",
        target_id=relationship.id,
        details={"relationship_type": relationship.relationship_type},
    )
    db.commit()
    db.refresh(relationship)
    return relationship


@router.get("/relationships", response_model=list[RelationshipRead])
def list_relationships(db: DB, limit: int = Query(default=100, ge=1, le=500)) -> list[Relationship]:
    return list(
        db.scalars(select(Relationship).order_by(Relationship.created_at.desc()).limit(limit)).all()
    )


@router.get("/entities/{entity_id}/graph", response_model=GraphRead)
def entity_graph(entity_id: str, db: DB) -> GraphRead:
    graph = graph_for_entity(db, entity_id)
    if graph is None:
        raise HTTPException(status_code=404, detail="Entity not found")
    return graph


@router.post("/workers/register", response_model=WorkerRegistered, status_code=status.HTTP_201_CREATED)
def register_worker(payload: WorkerRegister, db: DB, _admin: Admin) -> WorkerRegistered:
    token = issue_token()
    worker = Worker(
        name=payload.name,
        capabilities=payload.capabilities,
        token_hash=hash_token(token),
    )
    db.add(worker)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="worker.registered",
        target_type="worker",
        target_id=worker.id,
        details={"name": worker.name, "capabilities": worker.capabilities},
    )
    db.commit()
    db.refresh(worker)
    return WorkerRegistered(
        id=worker.id,
        name=worker.name,
        capabilities=worker.capabilities,
        token=token,
    )


@router.get("/workers", response_model=list[WorkerRead])
def list_workers(db: DB, _admin: Admin) -> list[Worker]:
    return list(db.scalars(select(Worker).order_by(Worker.created_at.desc())).all())


@router.post("/jobs", response_model=JobRead, status_code=status.HTTP_201_CREATED)
def submit_job(payload: JobCreate, db: DB, _admin: Admin) -> Job:
    if payload.task_type not in TASKS:
        raise HTTPException(
            status_code=400,
            detail=f"Unsupported task type. Supported: {', '.join(sorted(TASKS))}",
        )
    return create_job(db, **payload.model_dump())


@router.get("/jobs", response_model=list[JobRead])
def list_jobs(
    db: DB,
    _admin: Admin,
    limit: int = Query(default=100, ge=1, le=500),
) -> list[Job]:
    return list(
        db.scalars(
            select(Job)
            .options(selectinload(Job.work_units))
            .order_by(Job.created_at.desc())
            .limit(limit)
        ).all()
    )


@router.get("/jobs/{job_id}", response_model=JobRead)
def get_job(job_id: str, db: DB, _admin: Admin) -> Job:
    job = db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job_id)
    )
    if job is None:
        raise HTTPException(status_code=404, detail="Job not found")
    return job


@router.post("/work/claim", response_model=WorkClaim | None)
def claim_work(db: DB, worker: AuthenticatedWorker, response: Response) -> WorkClaim | None:
    unit = lease_work(db, worker)
    if unit is None:
        response.status_code = status.HTTP_204_NO_CONTENT
        return None
    return WorkClaim(
        work_unit_id=unit.id,
        lease_token=unit.lease_token or "",
        lease_expires_at=unit.lease_expires_at,
        job_id=unit.job.id,
        task_type=unit.job.task_type,
        input_payload=unit.job.input_payload,
    )


@router.post("/work/{work_unit_id}/submit", response_model=JobRead)
def submit_work_result(
    work_unit_id: str,
    payload: WorkSubmit,
    db: DB,
    worker: AuthenticatedWorker,
) -> Job:
    unit = db.get(WorkUnit, work_unit_id)
    if unit is None:
        raise HTTPException(status_code=404, detail="Work unit not found")
    try:
        return submit_result(
            db,
            unit=unit,
            worker=worker,
            lease_token=payload.lease_token,
            result_payload=payload.result_payload,
        )
    except ValueError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@router.post("/work/{work_unit_id}/fail", response_model=JobRead)
def report_work_failure(
    work_unit_id: str,
    payload: WorkFailure,
    db: DB,
    worker: AuthenticatedWorker,
) -> Job:
    unit = db.get(WorkUnit, work_unit_id)
    if unit is None:
        raise HTTPException(status_code=404, detail="Work unit not found")
    try:
        return fail_work(
            db,
            unit=unit,
            worker=worker,
            lease_token=payload.lease_token,
            error=payload.error,
        )
    except ValueError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@router.get("/audit", response_model=list[AuditRead])
def list_audit_events(
    db: DB,
    _admin: Admin,
    limit: int = Query(default=100, ge=1, le=500),
) -> list[AuditEvent]:
    return list(
        db.scalars(select(AuditEvent).order_by(AuditEvent.created_at.desc()).limit(limit)).all()
    )
OIG_FILE_025_ce65a5ba

mkdir -p 'src/oig'
cat > 'src/oig/config.py' <<'OIG_FILE_026_2efe1290'
from functools import lru_cache

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="OIG_",
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    app_name: str = "Open Intelligence Grid"
    environment: str = "development"
    database_url: str = "sqlite:///./oig.db"
    admin_key: str = "change-me-now"
    lease_seconds: int = 300
    allow_same_worker_replicas: bool = False
    cors_origins: list[str] = Field(default_factory=list)


@lru_cache
def get_settings() -> Settings:
    return Settings()
OIG_FILE_026_2efe1290

mkdir -p 'src/oig'
cat > 'src/oig/db.py' <<'OIG_FILE_027_ca680816'
from collections.abc import Generator

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

from .config import get_settings


class Base(DeclarativeBase):
    pass


_settings = get_settings()
_connect_args = {"check_same_thread": False} if _settings.database_url.startswith("sqlite") else {}
engine = create_engine(_settings.database_url, connect_args=_connect_args, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)


def init_db() -> None:
    from . import models  # noqa: F401

    Base.metadata.create_all(bind=engine)


def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
OIG_FILE_027_ca680816

mkdir -p 'src/oig'
cat > 'src/oig/main.py' <<'OIG_FILE_028_3aa66690'
from contextlib import asynccontextmanager

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse

from .api import router
from .config import get_settings
from .db import init_db


DASHBOARD = """
<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Open Intelligence Grid</title>
  <style>
    :root { color-scheme: dark; font-family: ui-monospace, SFMono-Regular, Menlo, monospace; }
    body { margin: 0; background: #101419; color: #e8f0e8; }
    main { max-width: 960px; margin: 0 auto; padding: 48px 24px; }
    h1 { font-size: clamp(2rem, 6vw, 4.5rem); margin: 0; letter-spacing: -0.06em; }
    .subtitle { color: #9db0a2; max-width: 720px; line-height: 1.6; }
    .grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(150px, 1fr)); gap: 12px; margin: 32px 0; }
    .card { border: 1px solid #334139; border-radius: 12px; padding: 18px; background: #151c19; }
    .number { font-size: 2rem; font-weight: 700; }
    a { color: #a9ffbe; }
    pre { white-space: pre-wrap; word-break: break-word; background: #0b0e0c; padding: 16px; border-radius: 12px; }
    .pulse { display: inline-block; width: 10px; height: 10px; border-radius: 50%; background: #a9ffbe; box-shadow: 0 0 16px #a9ffbe; }
  </style>
</head>
<body>
<main>
  <div><span class="pulse"></span> alpha coordinator online</div>
  <h1>Open Intelligence Grid</h1>
  <p class="subtitle">An inspectable ontology and distributed-computing foundation. Small enough to understand. Large enough to begin.</p>
  <div class="grid" id="stats"><div class="card">Loading statistics...</div></div>
  <p><a href="/docs">Interactive API documentation</a> · <a href="/health">Health check</a></p>
  <h2>Supported worker tasks</h2>
  <pre id="tasks">Loading...</pre>
</main>
<script>
fetch('/v1/stats').then(r => r.json()).then(data => {
  const entries = [
    ['Object types', data.object_types], ['Entities', data.entities],
    ['Relationships', data.relationships], ['Workers', data.workers],
    ['Jobs', Object.values(data.jobs).reduce((a,b) => a+b, 0)]
  ];
  document.getElementById('stats').innerHTML = entries.map(([k,v]) => `<div class="card"><div class="number">${v}</div><div>${k}</div></div>`).join('');
  document.getElementById('tasks').textContent = data.supported_tasks.join('\n');
}).catch(error => {
  document.getElementById('stats').innerHTML = `<div class="card">Could not load stats: ${error}</div>`;
});
</script>
</body>
</html>
"""


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_db()
    yield


settings = get_settings()
app = FastAPI(
    title=settings.app_name,
    version="0.1.0",
    description="Open ontology and distributed volunteer-computing coordinator.",
    lifespan=lifespan,
)
if settings.cors_origins:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=False,
        allow_methods=["GET", "POST", "PATCH", "DELETE"],
        allow_headers=["Authorization", "Content-Type", "X-Admin-Key"],
    )
app.include_router(router)


@app.get("/", response_class=HTMLResponse, include_in_schema=False)
def dashboard() -> str:
    return DASHBOARD


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok", "version": "0.1.0", "environment": settings.environment}


def run() -> None:
    uvicorn.run("oig.main:app", host="0.0.0.0", port=8000, reload=False)


if __name__ == "__main__":
    run()
OIG_FILE_028_3aa66690

mkdir -p 'src/oig'
cat > 'src/oig/models.py' <<'OIG_FILE_029_37463999'
from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, JSON, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship

from .db import Base


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


def uuid_string() -> str:
    return str(uuid.uuid4())


class ObjectType(Base):
    __tablename__ = "object_types"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    name: Mapped[str] = mapped_column(String(120), unique=True, index=True)
    description: Mapped[str] = mapped_column(Text, default="")
    schema_json: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)

    entities: Mapped[list[Entity]] = relationship(back_populates="object_type")


class Entity(Base):
    __tablename__ = "entities"
    __table_args__ = (UniqueConstraint("object_type_id", "external_id", name="uq_entity_external"),)

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    object_type_id: Mapped[str] = mapped_column(ForeignKey("object_types.id"), index=True)
    external_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
    properties: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow)

    object_type: Mapped[ObjectType] = relationship(back_populates="entities")


class Relationship(Base):
    __tablename__ = "relationships"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    relationship_type: Mapped[str] = mapped_column(String(120), index=True)
    source_entity_id: Mapped[str] = mapped_column(ForeignKey("entities.id"), index=True)
    target_entity_id: Mapped[str] = mapped_column(ForeignKey("entities.id"), index=True)
    properties: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)

    source: Mapped[Entity] = relationship(foreign_keys=[source_entity_id])
    target: Mapped[Entity] = relationship(foreign_keys=[target_entity_id])


class Worker(Base):
    __tablename__ = "workers"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    name: Mapped[str] = mapped_column(String(160))
    token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True)
    capabilities: Mapped[list[str]] = mapped_column(JSON, default=list)
    trust_score: Mapped[float] = mapped_column(Float, default=1.0)
    active: Mapped[bool] = mapped_column(Boolean, default=True)
    last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)


class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    task_type: Mapped[str] = mapped_column(String(120), index=True)
    input_payload: Mapped[dict[str, Any]] = mapped_column(JSON)
    status: Mapped[str] = mapped_column(String(30), default="queued", index=True)
    replicas_required: Mapped[int] = mapped_column(Integer, default=2)
    max_replicas: Mapped[int] = mapped_column(Integer, default=3)
    result_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
    result_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow)

    work_units: Mapped[list[WorkUnit]] = relationship(
        back_populates="job", cascade="all, delete-orphan", order_by="WorkUnit.created_at"
    )


class WorkUnit(Base):
    __tablename__ = "work_units"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    job_id: Mapped[str] = mapped_column(ForeignKey("jobs.id"), index=True)
    replica_index: Mapped[int] = mapped_column(Integer)
    status: Mapped[str] = mapped_column(String(30), default="queued", index=True)
    worker_id: Mapped[str | None] = mapped_column(ForeignKey("workers.id"), nullable=True, index=True)
    lease_token: Mapped[str | None] = mapped_column(String(64), nullable=True)
    lease_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    result_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
    result_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    error: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    job: Mapped[Job] = relationship(back_populates="work_units")
    worker: Mapped[Worker | None] = relationship()


class AuditEvent(Base):
    __tablename__ = "audit_events"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    actor_type: Mapped[str] = mapped_column(String(40))
    actor_id: Mapped[str] = mapped_column(String(160))
    action: Mapped[str] = mapped_column(String(160), index=True)
    target_type: Mapped[str] = mapped_column(String(80))
    target_id: Mapped[str] = mapped_column(String(160))
    details: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, index=True)
OIG_FILE_029_37463999

mkdir -p 'src/oig'
cat > 'src/oig/schemas.py' <<'OIG_FILE_030_ee96d89a'
from datetime import datetime
from typing import Any

from pydantic import BaseModel, ConfigDict, Field, model_validator


class ORMModel(BaseModel):
    model_config = ConfigDict(from_attributes=True)


class ObjectTypeCreate(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    name: str = Field(min_length=1, max_length=120, pattern=r"^[A-Za-z][A-Za-z0-9_-]*$")
    description: str = ""
    schema_definition: dict[str, Any] = Field(default_factory=dict, alias="schema_json")


class ObjectTypeRead(ORMModel):
    id: str
    name: str
    description: str
    schema_definition: dict[str, Any] = Field(
        validation_alias="schema_json", serialization_alias="schema_json"
    )
    created_at: datetime


class EntityCreate(BaseModel):
    object_type_id: str
    external_id: str | None = None
    properties: dict[str, Any] = Field(default_factory=dict)


class EntityRead(ORMModel):
    id: str
    object_type_id: str
    external_id: str | None
    properties: dict[str, Any]
    created_at: datetime
    updated_at: datetime


class RelationshipCreate(BaseModel):
    relationship_type: str = Field(min_length=1, max_length=120)
    source_entity_id: str
    target_entity_id: str
    properties: dict[str, Any] = Field(default_factory=dict)


class RelationshipRead(ORMModel):
    id: str
    relationship_type: str
    source_entity_id: str
    target_entity_id: str
    properties: dict[str, Any]
    created_at: datetime


class GraphRead(BaseModel):
    center: EntityRead
    entities: list[EntityRead]
    relationships: list[RelationshipRead]


class WorkerRegister(BaseModel):
    name: str = Field(min_length=1, max_length=160)
    capabilities: list[str] = Field(default_factory=list)


class WorkerRegistered(BaseModel):
    id: str
    name: str
    capabilities: list[str]
    token: str


class WorkerRead(ORMModel):
    id: str
    name: str
    capabilities: list[str]
    trust_score: float
    active: bool
    last_seen_at: datetime | None
    created_at: datetime


class JobCreate(BaseModel):
    task_type: str = Field(min_length=1, max_length=120)
    input_payload: dict[str, Any]
    replicas_required: int = Field(default=2, ge=1, le=10)
    max_replicas: int = Field(default=3, ge=1, le=20)

    @model_validator(mode="after")
    def validate_replica_counts(self) -> "JobCreate":
        if self.max_replicas < self.replicas_required:
            raise ValueError("max_replicas must be greater than or equal to replicas_required")
        return self


class WorkUnitRead(ORMModel):
    id: str
    job_id: str
    replica_index: int
    status: str
    worker_id: str | None
    lease_expires_at: datetime | None
    result_hash: str | None
    error: str | None
    created_at: datetime
    completed_at: datetime | None


class JobRead(ORMModel):
    id: str
    task_type: str
    input_payload: dict[str, Any]
    status: str
    replicas_required: int
    max_replicas: int
    result_payload: dict[str, Any] | None
    result_hash: str | None
    created_at: datetime
    updated_at: datetime
    work_units: list[WorkUnitRead] = Field(default_factory=list)


class WorkClaim(BaseModel):
    work_unit_id: str
    lease_token: str
    lease_expires_at: datetime
    job_id: str
    task_type: str
    input_payload: dict[str, Any]


class WorkSubmit(BaseModel):
    lease_token: str
    result_payload: dict[str, Any]


class WorkFailure(BaseModel):
    lease_token: str
    error: str = Field(min_length=1, max_length=4000)


class AuditRead(ORMModel):
    id: int
    actor_type: str
    actor_id: str
    action: str
    target_type: str
    target_id: str
    details: dict[str, Any]
    created_at: datetime
OIG_FILE_030_ee96d89a

mkdir -p 'src/oig'
cat > 'src/oig/security.py' <<'OIG_FILE_031_5014c70e'
import hashlib
import secrets

from fastapi import Depends, Header, HTTPException, status
from sqlalchemy import select
from sqlalchemy.orm import Session

from .config import get_settings
from .db import get_db
from .models import Worker, utcnow


def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def issue_token() -> str:
    return secrets.token_urlsafe(32)


def require_admin(x_admin_key: str = Header(default="")) -> str:
    settings = get_settings()
    if not x_admin_key or not secrets.compare_digest(x_admin_key, settings.admin_key):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid admin key")
    return "admin"


def require_worker(
    authorization: str = Header(default=""), db: Session = Depends(get_db)
) -> Worker:
    scheme, _, token = authorization.partition(" ")
    if scheme.lower() != "bearer" or not token:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Bearer token required")

    worker = db.scalar(select(Worker).where(Worker.token_hash == hash_token(token)))
    if worker is None or not worker.active:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid worker token")

    worker.last_seen_at = utcnow()
    db.commit()
    db.refresh(worker)
    return worker
OIG_FILE_031_5014c70e

mkdir -p 'src/oig'
cat > 'src/oig/services.py' <<'OIG_FILE_032_c7ed04cc'
from __future__ import annotations

import hashlib
import json
import secrets
from collections import Counter
from datetime import datetime, timedelta, timezone
from typing import Any

from sqlalchemy import or_, select
from sqlalchemy.orm import Session, selectinload

from .config import get_settings
from .models import AuditEvent, Entity, Job, Relationship, Worker, WorkUnit, utcnow
from .schemas import GraphRead


def _as_utc(value: datetime) -> datetime:
    """Normalize timestamps returned by SQLite, which may drop timezone metadata."""
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)


def canonical_hash(payload: dict[str, Any]) -> str:
    encoded = json.dumps(
        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def audit(
    db: Session,
    *,
    actor_type: str,
    actor_id: str,
    action: str,
    target_type: str,
    target_id: str,
    details: dict[str, Any] | None = None,
) -> None:
    db.add(
        AuditEvent(
            actor_type=actor_type,
            actor_id=actor_id,
            action=action,
            target_type=target_type,
            target_id=target_id,
            details=details or {},
        )
    )


def create_job(db: Session, *, task_type: str, input_payload: dict[str, Any], replicas_required: int, max_replicas: int) -> Job:
    job = Job(
        task_type=task_type,
        input_payload=input_payload,
        replicas_required=replicas_required,
        max_replicas=max_replicas,
    )
    db.add(job)
    db.flush()
    for replica_index in range(replicas_required):
        db.add(WorkUnit(job_id=job.id, replica_index=replica_index))
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="job.created",
        target_type="job",
        target_id=job.id,
        details={"task_type": task_type, "replicas_required": replicas_required},
    )
    db.commit()
    return db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job.id)
    )


def _release_expired_leases(db: Session) -> None:
    now = utcnow()
    expired = db.scalars(
        select(WorkUnit).where(
            WorkUnit.status == "leased",
            WorkUnit.lease_expires_at.is_not(None),
            WorkUnit.lease_expires_at < now,
        )
    ).all()
    for unit in expired:
        unit.status = "queued"
        unit.worker_id = None
        unit.lease_token = None
        unit.lease_expires_at = None
    if expired:
        db.commit()


def lease_work(db: Session, worker: Worker) -> WorkUnit | None:
    settings = get_settings()
    _release_expired_leases(db)

    candidates = db.scalars(
        select(WorkUnit)
        .options(selectinload(WorkUnit.job))
        .where(WorkUnit.status == "queued")
        .order_by(WorkUnit.created_at)
    ).all()

    for unit in candidates:
        job = unit.job
        if job.status not in {"queued", "running"}:
            continue
        if worker.capabilities and job.task_type not in worker.capabilities:
            continue
        if not settings.allow_same_worker_replicas:
            already_worked = db.scalar(
                select(WorkUnit.id).where(
                    WorkUnit.job_id == job.id,
                    WorkUnit.worker_id == worker.id,
                )
            )
            if already_worked is not None:
                continue

        unit.status = "leased"
        unit.worker_id = worker.id
        unit.lease_token = secrets.token_urlsafe(24)
        unit.lease_expires_at = utcnow() + timedelta(seconds=settings.lease_seconds)
        job.status = "running"
        audit(
            db,
            actor_type="worker",
            actor_id=worker.id,
            action="work.leased",
            target_type="work_unit",
            target_id=unit.id,
            details={"job_id": job.id},
        )
        db.commit()
        db.refresh(unit)
        return unit
    return None


def _evaluate_job(db: Session, job: Job) -> None:
    completed = [unit for unit in job.work_units if unit.status == "completed" and unit.result_hash]
    counts = Counter(unit.result_hash for unit in completed)

    if counts:
        winning_hash, winning_count = counts.most_common(1)[0]
        if winning_count >= job.replicas_required:
            winning_unit = next(unit for unit in completed if unit.result_hash == winning_hash)
            job.status = "completed"
            job.result_hash = winning_hash
            job.result_payload = winning_unit.result_payload
            for unit in job.work_units:
                if unit.status == "queued":
                    unit.status = "cancelled"
            audit(
                db,
                actor_type="system",
                actor_id="consensus",
                action="job.completed",
                target_type="job",
                target_id=job.id,
                details={"result_hash": winning_hash, "matching_replicas": winning_count},
            )
            return

    active_or_waiting = [unit for unit in job.work_units if unit.status in {"queued", "leased"}]
    if active_or_waiting:
        return

    if len(job.work_units) < job.max_replicas:
        next_index = max((unit.replica_index for unit in job.work_units), default=-1) + 1
        job.work_units.append(WorkUnit(job_id=job.id, replica_index=next_index))
        job.status = "running"
        audit(
            db,
            actor_type="system",
            actor_id="consensus",
            action="work.tie_breaker_created",
            target_type="job",
            target_id=job.id,
            details={"replica_index": next_index},
        )
    else:
        job.status = "disputed"
        audit(
            db,
            actor_type="system",
            actor_id="consensus",
            action="job.disputed",
            target_type="job",
            target_id=job.id,
            details={"hash_counts": dict(counts)},
        )


def submit_result(
    db: Session,
    *,
    unit: WorkUnit,
    worker: Worker,
    lease_token: str,
    result_payload: dict[str, Any],
) -> Job:
    if unit.status != "leased" or unit.worker_id != worker.id:
        raise ValueError("Work unit is not leased to this worker")
    if not unit.lease_token or not secrets.compare_digest(unit.lease_token, lease_token):
        raise ValueError("Invalid lease token")
    if unit.lease_expires_at and _as_utc(unit.lease_expires_at) < utcnow():
        raise ValueError("Lease expired")

    unit.status = "completed"
    unit.result_payload = result_payload
    unit.result_hash = canonical_hash(result_payload)
    unit.completed_at = utcnow()
    unit.lease_token = None
    unit.lease_expires_at = None
    worker.trust_score = min(worker.trust_score + 0.01, 10.0)
    audit(
        db,
        actor_type="worker",
        actor_id=worker.id,
        action="work.completed",
        target_type="work_unit",
        target_id=unit.id,
        details={"result_hash": unit.result_hash},
    )

    db.flush()
    job = db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == unit.job_id)
    )
    if job is None:
        raise ValueError("Parent job not found")
    _evaluate_job(db, job)
    db.commit()
    return db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job.id)
    )


def fail_work(
    db: Session, *, unit: WorkUnit, worker: Worker, lease_token: str, error: str
) -> Job:
    if unit.status != "leased" or unit.worker_id != worker.id:
        raise ValueError("Work unit is not leased to this worker")
    if not unit.lease_token or not secrets.compare_digest(unit.lease_token, lease_token):
        raise ValueError("Invalid lease token")

    unit.status = "failed"
    unit.error = error
    unit.completed_at = utcnow()
    unit.lease_token = None
    unit.lease_expires_at = None
    worker.trust_score = max(worker.trust_score - 0.02, 0.0)

    db.flush()
    job = db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == unit.job_id)
    )
    if job is None:
        raise ValueError("Parent job not found")
    _evaluate_job(db, job)
    db.commit()
    return db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job.id)
    )


def graph_for_entity(db: Session, entity_id: str) -> GraphRead | None:
    center = db.get(Entity, entity_id)
    if center is None:
        return None
    relationships = db.scalars(
        select(Relationship).where(
            or_(
                Relationship.source_entity_id == entity_id,
                Relationship.target_entity_id == entity_id,
            )
        )
    ).all()
    neighbor_ids = {
        rel.target_entity_id if rel.source_entity_id == entity_id else rel.source_entity_id
        for rel in relationships
    }
    neighbors = db.scalars(select(Entity).where(Entity.id.in_(neighbor_ids))).all() if neighbor_ids else []
    return GraphRead(center=center, entities=[center, *neighbors], relationships=relationships)
OIG_FILE_032_c7ed04cc

mkdir -p 'src/oig'
cat > 'src/oig/tasks.py' <<'OIG_FILE_033_dc75af9d'
import hashlib
import json
from collections.abc import Callable
from typing import Any


class TaskError(ValueError):
    pass


def word_count(payload: dict[str, Any]) -> dict[str, Any]:
    text = payload.get("text")
    if not isinstance(text, str):
        raise TaskError("word_count requires a string field named 'text'")
    return {
        "words": len(text.split()),
        "characters": len(text),
        "lines": len(text.splitlines()) if text else 0,
    }


def sum_numbers(payload: dict[str, Any]) -> dict[str, Any]:
    numbers = payload.get("numbers")
    if not isinstance(numbers, list) or not all(
        isinstance(value, int) and not isinstance(value, bool) for value in numbers
    ):
        raise TaskError("sum_numbers requires an integer list field named 'numbers'")
    return {"sum": sum(numbers), "count": len(numbers)}


def hash_text(payload: dict[str, Any]) -> dict[str, Any]:
    text = payload.get("text")
    algorithm = payload.get("algorithm", "sha256")
    if not isinstance(text, str):
        raise TaskError("hash_text requires a string field named 'text'")
    if algorithm not in {"sha256", "sha512", "blake2b"}:
        raise TaskError("algorithm must be sha256, sha512, or blake2b")
    digest = hashlib.new(algorithm, text.encode("utf-8")).hexdigest()
    return {"algorithm": algorithm, "digest": digest}


def json_stats(payload: dict[str, Any]) -> dict[str, Any]:
    def walk(value: Any) -> tuple[int, int, int, int]:
        if isinstance(value, dict):
            child = [walk(item) for item in value.values()]
            return (
                1 + sum(item[0] for item in child),
                sum(item[1] for item in child),
                sum(item[2] for item in child),
                sum(item[3] for item in child),
            )
        if isinstance(value, list):
            child = [walk(item) for item in value]
            return (
                sum(item[0] for item in child),
                1 + sum(item[1] for item in child),
                sum(item[2] for item in child),
                sum(item[3] for item in child),
            )
        if isinstance(value, (str, int, float, bool)) or value is None:
            return (0, 0, 1, len(json.dumps(value, ensure_ascii=False)))
        raise TaskError("payload contains an unsupported JSON value")

    objects, arrays, scalars, scalar_bytes = walk(payload)
    return {
        "objects": objects,
        "arrays": arrays,
        "scalars": scalars,
        "serialized_scalar_characters": scalar_bytes,
    }


TASKS: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = {
    "word_count": word_count,
    "sum_numbers": sum_numbers,
    "hash_text": hash_text,
    "json_stats": json_stats,
}


def run_task(task_type: str, payload: dict[str, Any]) -> dict[str, Any]:
    task = TASKS.get(task_type)
    if task is None:
        raise TaskError(f"Unsupported task type: {task_type}")
    return task(payload)
OIG_FILE_033_dc75af9d

mkdir -p 'src/oig'
cat > 'src/oig/worker.py' <<'OIG_FILE_034_799e1e06'
import argparse
import sys
import time

import httpx

from .tasks import TaskError, run_task


def worker_headers(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def process_once(server: str, token: str, timeout: float = 30.0) -> bool:
    server = server.rstrip("/")
    with httpx.Client(timeout=timeout, headers=worker_headers(token)) as client:
        claim = client.post(f"{server}/v1/work/claim")
        if claim.status_code == 204:
            return False
        claim.raise_for_status()
        work = claim.json()

        try:
            result = run_task(work["task_type"], work["input_payload"])
        except (TaskError, Exception) as exc:
            failure = client.post(
                f"{server}/v1/work/{work['work_unit_id']}/fail",
                json={"lease_token": work["lease_token"], "error": str(exc)},
            )
            failure.raise_for_status()
            print(f"Failed work unit {work['work_unit_id']}: {exc}", file=sys.stderr)
            return True

        response = client.post(
            f"{server}/v1/work/{work['work_unit_id']}/submit",
            json={"lease_token": work["lease_token"], "result_payload": result},
        )
        response.raise_for_status()
        job = response.json()
        print(
            f"Completed work unit {work['work_unit_id']} for job {work['job_id']} "
            f"(job status: {job['status']})"
        )
        return True


def main() -> None:
    parser = argparse.ArgumentParser(description="Open Intelligence Grid volunteer worker")
    parser.add_argument("--server", default="http://localhost:8000")
    parser.add_argument("--token", required=True, help="Worker token returned during registration")
    parser.add_argument("--once", action="store_true", help="Claim at most one work unit and exit")
    parser.add_argument("--poll-seconds", type=float, default=10.0)
    args = parser.parse_args()

    while True:
        try:
            worked = process_once(args.server, args.token)
        except httpx.HTTPError as exc:
            print(f"Coordinator request failed: {exc}", file=sys.stderr)
            worked = False

        if args.once:
            return
        if not worked:
            time.sleep(max(args.poll_seconds, 1.0))


if __name__ == "__main__":
    main()
OIG_FILE_034_799e1e06

mkdir -p 'tests'
cat > 'tests/conftest.py' <<'OIG_FILE_035_626af34e'
from pathlib import Path

import pytest
from fastapi.testclient import TestClient


@pytest.fixture()
def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    database_path = tmp_path / "test.db"
    monkeypatch.setenv("OIG_DATABASE_URL", f"sqlite:///{database_path}")
    monkeypatch.setenv("OIG_ADMIN_KEY", "test-admin")
    monkeypatch.setenv("OIG_ALLOW_SAME_WORKER_REPLICAS", "false")

    import oig.config as config

    config.get_settings.cache_clear()

    import oig.db as db

    db.engine.dispose()
    connect_args = {"check_same_thread": False}
    db.engine = db.create_engine(
        f"sqlite:///{database_path}", connect_args=connect_args, pool_pre_ping=True
    )
    db.SessionLocal.configure(bind=db.engine)

    from oig.main import app

    with TestClient(app) as test_client:
        yield test_client
OIG_FILE_035_626af34e

mkdir -p 'tests'
cat > 'tests/test_health.py' <<'OIG_FILE_036_fe7062cb'
def test_health(client):
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "ok"
OIG_FILE_036_fe7062cb

mkdir -p 'tests'
cat > 'tests/test_jobs.py' <<'OIG_FILE_037_9cb68222'
ADMIN = {"X-Admin-Key": "test-admin"}


def register_worker(client, name):
    response = client.post(
        "/v1/workers/register",
        headers=ADMIN,
        json={"name": name, "capabilities": ["sum_numbers"]},
    )
    assert response.status_code == 201
    return response.json()


def claim(client, token):
    response = client.post(
        "/v1/work/claim", headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 200
    return response.json()


def submit(client, token, work, result):
    return client.post(
        f"/v1/work/{work['work_unit_id']}/submit",
        headers={"Authorization": f"Bearer {token}"},
        json={"lease_token": work["lease_token"], "result_payload": result},
    )


def test_two_workers_reach_consensus(client):
    worker_a = register_worker(client, "a")
    worker_b = register_worker(client, "b")

    job = client.post(
        "/v1/jobs",
        headers=ADMIN,
        json={
            "task_type": "sum_numbers",
            "input_payload": {"numbers": [1, 2, 3]},
            "replicas_required": 2,
            "max_replicas": 3,
        },
    )
    assert job.status_code == 201

    work_a = claim(client, worker_a["token"])
    work_b = claim(client, worker_b["token"])

    result_a = submit(client, worker_a["token"], work_a, {"sum": 6, "count": 3})
    result_b = submit(client, worker_b["token"], work_b, {"count": 3, "sum": 6})
    assert result_a.status_code == 200
    assert result_b.status_code == 200
    assert result_b.json()["status"] == "completed"
    assert result_b.json()["result_payload"] == {"count": 3, "sum": 6}


def test_disagreement_creates_tie_breaker(client):
    workers = [register_worker(client, name) for name in ("a", "b", "c")]
    client.post(
        "/v1/jobs",
        headers=ADMIN,
        json={
            "task_type": "sum_numbers",
            "input_payload": {"numbers": [4, 5]},
            "replicas_required": 2,
            "max_replicas": 3,
        },
    ).json()

    first = claim(client, workers[0]["token"])
    second = claim(client, workers[1]["token"])
    submit(client, workers[0]["token"], first, {"sum": 9, "count": 2})
    disputed_so_far = submit(client, workers[1]["token"], second, {"sum": 99, "count": 2})
    assert disputed_so_far.status_code == 200
    assert len(disputed_so_far.json()["work_units"]) == 3

    tie_breaker = claim(client, workers[2]["token"])
    final = submit(client, workers[2]["token"], tie_breaker, {"count": 2, "sum": 9})
    assert final.status_code == 200
    assert final.json()["status"] == "completed"
    assert final.json()["result_payload"]["sum"] == 9
OIG_FILE_037_9cb68222

mkdir -p 'tests'
cat > 'tests/test_ontology.py' <<'OIG_FILE_038_a47b6a15'
ADMIN = {"X-Admin-Key": "test-admin"}


def test_create_entities_and_graph(client):
    org_type = client.post(
        "/v1/object-types",
        headers=ADMIN,
        json={"name": "Organization", "schema_json": {"type": "object"}},
    )
    assert org_type.status_code == 201

    program_type = client.post(
        "/v1/object-types",
        headers=ADMIN,
        json={"name": "Program", "schema_json": {"type": "object"}},
    )
    assert program_type.status_code == 201

    org = client.post(
        "/v1/entities",
        headers=ADMIN,
        json={"object_type_id": org_type.json()["id"], "properties": {"name": "Library"}},
    )
    program = client.post(
        "/v1/entities",
        headers=ADMIN,
        json={"object_type_id": program_type.json()["id"], "properties": {"name": "Robotics"}},
    )
    assert org.status_code == 201
    assert program.status_code == 201

    relation = client.post(
        "/v1/relationships",
        headers=ADMIN,
        json={
            "relationship_type": "OPERATES",
            "source_entity_id": org.json()["id"],
            "target_entity_id": program.json()["id"],
            "properties": {},
        },
    )
    assert relation.status_code == 201

    graph = client.get(f"/v1/entities/{org.json()['id']}/graph")
    assert graph.status_code == 200
    assert len(graph.json()["entities"]) == 2
    assert graph.json()["relationships"][0]["relationship_type"] == "OPERATES"
OIG_FILE_038_a47b6a15

echo "Done. Next steps:"
echo "  python -m venv .venv"
echo "  source .venv/bin/activate"
echo "  pip install -e .[dev]"
echo "  pytest"
echo "  uvicorn oig.main:app --reload"

 

 

Here is the MD:

# Open Intelligence Grid v0.1.0 — Complete Cut-and-Paste Source Code

This document contains every source, config, test, and documentation file needed to recreate the Open Intelligence Grid v0.1.0 repository. Copy individual file blocks into matching paths, or use the companion shell script to recreate the project automatically.

## Repository tree

```text
.dockerignore
.env.example
.github/workflows/ci.yml
.gitignore
CHANGELOG.md
CODE_OF_CONDUCT.md
CONTRIBUTING.md
Dockerfile
GOVERNANCE.md
LICENSE
Makefile
README.md
SECURITY.md
compose.yaml
docs/API.md
docs/ARCHITECTURE.md
docs/GITHUB_PUBLISHING.md
docs/QUICKSTART.md
docs/RELEASE_VALIDATION.md
docs/ROADMAP.md
docs/THREAT_MODEL.md
examples/bootstrap_demo.py
pyproject.toml
src/oig/__init__.py
src/oig/api.py
src/oig/config.py
src/oig/db.py
src/oig/main.py
src/oig/models.py
src/oig/schemas.py
src/oig/security.py
src/oig/services.py
src/oig/tasks.py
src/oig/worker.py
tests/conftest.py
tests/test_health.py
tests/test_jobs.py
tests/test_ontology.py
```

## Recommended quick recreation

Use the companion script `create-open-intelligence-grid.sh`, or manually create the files below.

## Complete files

### `.dockerignore`

```gitignore
.git
.venv
__pycache__
.pytest_cache
*.db
.env
```

### `.env.example`

```bash
OIG_APP_NAME=Open Intelligence Grid
OIG_ENVIRONMENT=development
OIG_DATABASE_URL=sqlite:///./oig.db
OIG_ADMIN_KEY=change-me-now
OIG_LEASE_SECONDS=300
OIG_ALLOW_SAME_WORKER_REPLICAS=false
OIG_CORS_ORIGINS=[]
```

### `.github/workflows/ci.yml`

```yaml
name: CI

on:
  push:
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.11", "3.12", "3.13"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
      - run: python -m pip install --upgrade pip
      - run: pip install -e '.[dev]'
      - run: ruff check src tests
      - run: pytest
```

### `.gitignore`

```gitignore
.venv/
__pycache__/
*.py[cod]
.pytest_cache/
.ruff_cache/
.coverage
htmlcov/
.env
*.db
*.sqlite3
dist/
build/
*.egg-info/
.DS_Store
```

### `CHANGELOG.md`

```markdown
# Changelog

## 0.1.0 - 2026-07-01

### Added

- Ontology object types, entities, and relationships.
- One-hop graph traversal.
- Worker registration and bearer-token authentication.
- Database-backed work leasing.
- Replicated deterministic jobs with hash-based consensus.
- Tie-breaking work units for conflicting results.
- SQLite and PostgreSQL support.
- Docker Compose environment.
- Minimal browser dashboard.
- Tests and initial security documentation.
```

### `CODE_OF_CONDUCT.md`

```markdown
# Community code of conduct

## Our standard

Participants should help create a technically serious, curious, and humane community. Expected behavior includes:

- Critiquing ideas and code without degrading people.
- Welcoming contributors with different levels of experience.
- Documenting decisions that affect users or communities.
- Respecting privacy, consent, and the right to correct errors.
- Disclosing conflicts of interest.
- Avoiding harassment, threats, discrimination, and deliberate disruption.

## Public-interest responsibility

This project can become powerful. Contributors must not use project spaces to promote covert surveillance, harassment, discriminatory targeting, or collection of personal information without a lawful and ethical basis.

## Enforcement

Before public launch, maintainers should publish a dedicated conduct contact, a private reporting method, an appeal procedure, and a transparent range of enforcement actions.
```

### `CONTRIBUTING.md`

````markdown
# Contributing

Thank you for helping build an inspectable public-interest data platform.

## Development setup

```bash
python -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]'
pytest
ruff check src tests
```

## Contribution rules

1. Keep pull requests focused and documented.
2. Add tests for behavior changes.
3. Do not add surveillance-specific features without a public-interest review.
4. Do not add dependencies without explaining why the standard library or an existing dependency is insufficient.
5. Never commit secrets or real personal data.
6. Preserve backward compatibility or document a migration path.

## Commit suggestions

Use clear prefixes such as `feat:`, `fix:`, `docs:`, `test:`, and `security:`.

## Definition of done

A change should include implementation, tests, documentation, and a note in `CHANGELOG.md` when users will notice it.
````

### `Dockerfile`

```dockerfile
FROM python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

WORKDIR /app

COPY pyproject.toml README.md LICENSE ./
COPY src ./src
RUN pip install --no-cache-dir .

EXPOSE 8000
CMD ["uvicorn", "oig.main:app", "--host", "0.0.0.0", "--port", "8000"]
```

### `GOVERNANCE.md`

```markdown
# Governance draft

OIG should eventually be governed by a nonprofit or multistakeholder foundation rather than a single vendor.

## Proposed bodies

- **Maintainers:** accept code and manage releases.
- **Security team:** coordinates private vulnerability reports.
- **Public-interest council:** reviews high-risk capabilities and deployment patterns.
- **Data-governance working group:** develops consent, retention, provenance, and redress standards.
- **Community assembly:** proposes roadmap items and elects or confirms council members.

## High-risk change review

Features involving identity resolution, location tracking, biometric inference, predictive policing, mass scraping, or covert monitoring should require an explicit written review and documented safeguards before merging.

This is a starting constitution, not stone tablets descending from a venture-capital cloud.
```

### `LICENSE`

```text
MIT License

Copyright (c) 2026 Open Intelligence Grid contributors

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
```

### `Makefile`

```makefile
.PHONY: install dev test lint run docker-up docker-down

install:
python -m pip install -e .

dev:
python -m pip install -e '.[dev]'

test:
pytest

lint:
ruff check src tests

run:
oig-server

docker-up:
docker compose up --build

docker-down:
docker compose down
```

### `README.md`

````markdown
# Open Intelligence Grid

**Open Intelligence Grid (OIG)** is an early, working open-source foundation for a transparent data-and-operations platform with distributed volunteer computing.

It combines two ideas:

1. An **ontology layer** that represents real-world object types, entities, and relationships.
2. A **coordinator/worker network** that divides deterministic jobs into independently validated work units.

This is a usable alpha, not a production replacement for Palantir Foundry. It is intentionally small enough to understand, run, audit, and improve.

## What works in v0.1.0

- Define ontology object types with JSON Schema-like property descriptions.
- Create entities and relationships.
- Traverse a one-hop entity graph.
- Register authenticated volunteer workers.
- Submit distributed jobs with configurable replication.
- Lease work units to separate workers.
- Validate results by canonical-hash consensus.
- Add a tie-breaking replica when workers disagree.
- Record basic audit events.
- Run on SQLite for development or PostgreSQL with Docker Compose.
- Use the bundled deterministic worker tasks:
  - `word_count`
  - `sum_numbers` (integer inputs)
  - `hash_text`
  - `json_stats`

## Safety boundary

Do **not** put secrets, personal records, medical records, private messages, credentials, or other sensitive data into volunteer-computing jobs. Workers must be treated as untrusted machines. OIG v0.1.0 is for public, synthetic, sanitized, or otherwise safe-to-distribute inputs.

Read [SECURITY.md](SECURITY.md) and [docs/THREAT_MODEL.md](docs/THREAT_MODEL.md) before exposing a deployment to the internet.

## Quick start with Python

Requirements: Python 3.11 or newer.

```bash
python -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]'
cp .env.example .env
oig-server
```

Open:

- API documentation: `http://localhost:8000/docs`
- Minimal dashboard: `http://localhost:8000/`
- Health check: `http://localhost:8000/health`

The development admin key in `.env.example` is `change-me-now`. Change it before doing anything beyond local testing.

## Quick start with Docker

```bash
cp .env.example .env
docker compose up --build
```

The Compose setup starts PostgreSQL and the API.

## Run the demo

With the server running:

```bash
python examples/bootstrap_demo.py
```

The script creates an ontology, two workers, and a replicated word-count job. It prints two worker tokens. In two terminals, run:

```bash
oig-worker --server http://localhost:8000 --token WORKER_ONE_TOKEN --once
oig-worker --server http://localhost:8000 --token WORKER_TWO_TOKEN --once
```

Then inspect the job:

```bash
curl http://localhost:8000/v1/jobs/JOB_ID \
  -H 'X-Admin-Key: change-me-now'
```

## Repository map

```text
src/oig/
  api.py          HTTP routes
  config.py       Environment-based settings
  db.py           SQLAlchemy engine/session setup
  models.py       Persistent ontology, jobs, workers, and audit records
  schemas.py      Request and response models
  security.py     Admin and worker authentication
  services.py     Leasing, consensus, and graph services
  tasks.py        Safe deterministic worker task implementations
  worker.py       Volunteer worker client
  main.py         Application entry point and dashboard

tests/            API and consensus tests
docs/             Architecture, API, threat model, and roadmap
examples/         Runnable bootstrap example
```

## Guiding principles

- Public-interest uses before surveillance uses.
- Data minimization before data accumulation.
- Traceable transformations before black boxes.
- Compute moves toward sensitive data; sensitive data does not move toward strangers.
- No cryptocurrency is required for participation.
- Human governance is part of the architecture, not an afterthought.

## Current limitations

- The scheduler uses a relational database, not a peer-to-peer mesh.
- API-key authentication is suitable only for development and controlled pilots.
- There is no row-level policy engine yet.
- There are no signed job bundles or WebAssembly sandboxes yet.
- The ontology schema is stored and documented but not fully JSON-Schema validated.
- Database migrations are not yet included.
- Work leasing is deliberately simple and needs stronger transaction isolation at scale.

These limits are documented rather than hidden behind a cloud of enterprise incense.

## License

MIT. See [LICENSE](LICENSE).

## Working title

“Open Intelligence Grid” is a descriptive working title. Perform a proper trademark and project-name review before a major public launch.
````

### `SECURITY.md`

```markdown
# Security policy

## Status

OIG v0.1.0 is an alpha research and development release. It has not received a professional security audit and must not be used as-is for classified, regulated, medical, financial, or personally identifying information.

## Non-negotiable data rule

Volunteer workers are untrusted. Never distribute secrets or sensitive source records to them. Encryption in transit does not change this rule because a worker must ordinarily access a task's plaintext input to calculate its output.

## Supported deployment posture

For the alpha release:

- Keep the API behind a trusted reverse proxy.
- Use TLS.
- Replace the development admin key.
- Restrict administrative endpoints by network and identity.
- Use PostgreSQL for shared deployments.
- Log and monitor worker registrations and job submissions.
- Use only public, synthetic, or sanitized job inputs.
- Run trusted data processing on organization-controlled workers.

## Reporting vulnerabilities

Do not open a public issue containing exploit details or exposed data. Before public launch, create a dedicated security contact and private disclosure channel in the repository's `SECURITY.md`.

## Planned controls

See [docs/ROADMAP.md](docs/ROADMAP.md) for signed job bundles, scoped identities, policy enforcement, sandboxed execution, provenance signatures, and federation.
```

### `compose.yaml`

```yaml
services:
  db:
    image: postgres:17-alpine
    environment:
      POSTGRES_DB: oig
      POSTGRES_USER: oig
      POSTGRES_PASSWORD: oig-development-password
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U oig -d oig"]
      interval: 5s
      timeout: 5s
      retries: 10
    volumes:
      - oig-postgres:/var/lib/postgresql/data

  api:
    build: .
    depends_on:
      db:
        condition: service_healthy
    environment:
      OIG_ENVIRONMENT: development
      OIG_DATABASE_URL: postgresql+psycopg://oig:oig-development-password@db:5432/oig
      OIG_ADMIN_KEY: ${OIG_ADMIN_KEY:-change-me-now}
      OIG_LEASE_SECONDS: ${OIG_LEASE_SECONDS:-300}
      OIG_ALLOW_SAME_WORKER_REPLICAS: ${OIG_ALLOW_SAME_WORKER_REPLICAS:-false}
    ports:
      - "8000:8000"

volumes:
  oig-postgres:
```

### `docs/API.md`

````markdown
# API overview

Interactive OpenAPI documentation is available at `/docs` while the server is running.

## Authentication

Administrative endpoints require:

```http
X-Admin-Key: your-admin-key
```

Worker endpoints require:

```http
Authorization: Bearer worker-token
```

## Important endpoints

| Method | Path | Purpose |
|---|---|---|
| GET | `/health` | Liveness check |
| GET | `/v1/stats` | Counts and job status summary |
| POST | `/v1/object-types` | Create an ontology object type |
| POST | `/v1/entities` | Create an entity |
| POST | `/v1/relationships` | Create a relationship |
| GET | `/v1/entities/{id}/graph` | Read a one-hop graph |
| POST | `/v1/workers/register` | Register a worker and receive its token |
| POST | `/v1/jobs` | Create a replicated job |
| POST | `/v1/work/claim` | Lease work to an authenticated worker |
| POST | `/v1/work/{id}/submit` | Submit a completed result |
| POST | `/v1/work/{id}/fail` | Report a failed work unit |

## Create a job

```bash
curl -X POST http://localhost:8000/v1/jobs \
  -H 'Content-Type: application/json' \
  -H 'X-Admin-Key: change-me-now' \
  -d '{
    "task_type": "sum_numbers",
    "input_payload": {"numbers": [1, 2, 3, 4]},
    "replicas_required": 2,
    "max_replicas": 3
  }'
```
````

### `docs/ARCHITECTURE.md`

````markdown
# Architecture

## Current shape

```text
                         +-----------------------+
                         |   Browser / API user  |
                         +-----------+-----------+
                                     |
                                     v
+----------------+        +----------+-----------+        +----------------+
| Volunteer      | <----> | OIG Coordinator API | <----> | PostgreSQL or  |
| Worker A       |  HTTPS | FastAPI + services  |  SQL   | SQLite         |
+----------------+        +----------+-----------+        +----------------+
                                     ^
+----------------+                   |
| Volunteer      | <-----------------+
| Worker B       |
+----------------+
```

The coordinator is centralized in v0.1.0. Workers poll for jobs, lease a work unit, execute a known deterministic task, and submit a result. The coordinator accepts a job when enough independent workers produce the same canonical result hash.

## Ontology model

- `ObjectType` describes a class of real-world object and its expected properties.
- `Entity` is an instance of an object type.
- `Relationship` is a directed, typed edge between two entities.

The alpha stores property descriptions and entity properties as JSON. A later release should validate values against JSON Schema and add versioned ontology migrations.

## Job model

- `Job` contains a task type, input payload, validation threshold, and result.
- `WorkUnit` is one independently executed replica of a job.
- `Worker` has a bearer token, capabilities, activity state, and simple trust score.

A worker ordinarily cannot receive two replicas of the same job. Development deployments may override this with `OIG_ALLOW_SAME_WORKER_REPLICAS=true`.

## Consensus

Results are serialized as canonical JSON and hashed with SHA-256. A job completes when one hash has at least `replicas_required` completed replicas. If all issued replicas finish without consensus and the job has not reached `max_replicas`, the coordinator adds a tie-breaker work unit. Otherwise, the job becomes `disputed`.

This detects accidental and some malicious errors. It does not prove that a majority of workers are honest.

## Evolution path

1. Relational coordinator with trusted HTTPS workers.
2. Signed task manifests and sandboxed WebAssembly execution.
3. Organization-operated trusted worker pools.
4. Federated query and model execution near private data.
5. Replicated coordinators and append-only provenance logs.
6. Optional peer discovery and content-addressed task distribution.

Mesh networking belongs near the end because decentralizing an unclear protocol merely distributes the confusion.
````

### `docs/GITHUB_PUBLISHING.md`

````markdown
# Publishing this repository on GitHub

## 1. Review the working title

“Open Intelligence Grid” is a working title. Search existing projects and trademarks before promoting it as a permanent brand.

## 2. Create an empty repository

Create a new repository without automatically adding a README, license, or `.gitignore`, because those files already exist here.

## 3. Initialize and push

```bash
git init
git add .
git commit -m "Initial open-source alpha"
git branch -M main
git remote add origin YOUR_REPOSITORY_URL
git push -u origin main
```

## 4. Configure repository protections

Recommended settings:

- Require pull requests before merging to `main`.
- Require the CI workflow to pass.
- Require at least one approving review.
- Block force pushes and branch deletion.
- Enable secret scanning and dependency alerts where available.
- Create a private security-reporting channel before inviting broad use.

## 5. Create the first release

Tag the tested alpha:

```bash
git tag -a v0.1.0 -m "Open Intelligence Grid alpha"
git push origin v0.1.0
```

Attach the source archive and describe the current safety boundaries prominently. Do not present the alpha as suitable for sensitive or regulated data.

## 6. Suggested repository description

> Open-source ontology, data coordination, and independently validated volunteer-computing foundation.

## 7. Suggested initial topics

`open-source`, `distributed-computing`, `volunteer-computing`, `ontology`, `fastapi`, `postgresql`, `civic-tech`, `data-governance`
````

### `docs/QUICKSTART.md`

````markdown
# Quickstart

## 1. Install

```bash
python -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]'
cp .env.example .env
```

## 2. Start the coordinator

```bash
oig-server
```

## 3. Bootstrap a demonstration

```bash
python examples/bootstrap_demo.py
```

Copy the two printed worker tokens and job ID.

## 4. Run two independent workers

Terminal A:

```bash
oig-worker --server http://localhost:8000 --token TOKEN_A --once
```

Terminal B:

```bash
oig-worker --server http://localhost:8000 --token TOKEN_B --once
```

## 5. Read the accepted job

```bash
curl http://localhost:8000/v1/jobs/JOB_ID \
  -H 'X-Admin-Key: change-me-now'
```

The job should have status `completed` and two matching completed work units.
````

### `docs/RELEASE_VALIDATION.md`

````markdown
# Release validation

Release: `0.1.0`

Validation performed on July 1, 2026 with Python 3.13.

## Automated checks

```text
pytest: 4 passed
ruff: all checks passed
compileall: passed
compose.yaml parse: passed
editable package installation: passed
```

## End-to-end check

A local coordinator was started with SQLite. The demonstration script then:

1. Created an ontology object type and entity.
2. Registered two distinct workers.
3. Created a word-count job requiring two matching replicas.
4. Ran each worker once through the command-line client.
5. Received two identical canonical result hashes.
6. Marked the job `completed` with the verified result.

## Important scope statement

These checks demonstrate that the alpha functions as designed in a local development environment. They are not a security audit, load test, formal protocol verification, or production-readiness certification.
````

### `docs/ROADMAP.md`

```markdown
# Roadmap

## Phase 1: Harden the alpha

- Alembic database migrations.
- Pagination and filtering.
- Idempotency keys.
- Rate limiting.
- Structured logs, metrics, and traces.
- Stronger transaction-safe work claiming.
- JSON Schema validation for ontology entities.
- Capability and resource matching.
- Worker heartbeat and health screens.

## Phase 2: Safe distributed execution

- Signed task manifests.
- Content-addressed input bundles.
- WebAssembly/WASI task sandbox.
- CPU, memory, disk, and time quotas.
- Network-disabled tasks by default.
- Reproducible task builds.
- Random known-answer challenge jobs.
- Weighted consensus and worker reputation.

## Phase 3: Institutional federation

- Organizations, projects, and scoped roles.
- OpenID Connect authentication.
- Open Policy Agent integration.
- Data residency policies.
- Trusted worker pools.
- Federated aggregate queries.
- Differential privacy options.
- Signed lineage and provenance export.

## Phase 4: Operational applications

- Ontology actions and approval workflows.
- Event streams and alerting.
- Geospatial object support.
- Notebook and SQL workspaces.
- Visual pipeline builder.
- Dashboard/application SDK.
- Connectors for public datasets and common databases.

## Phase 5: Resilient network

- Replicated coordinators.
- Append-only transparent audit log.
- Peer discovery.
- Content-addressed peer-to-peer distribution for public task bundles.
- Offline-capable edge nodes.
- Byzantine-fault research and formal protocol specification.

A full Palantir-class platform is a long-range ecosystem. Each phase must remain independently useful rather than serving as scaffolding for a castle that never opens its doors.
```

### `docs/THREAT_MODEL.md`

```markdown
# Threat model

## Assets

- Ontology definitions and entity data.
- Administrative credentials.
- Worker identities and tokens.
- Job inputs and accepted results.
- Audit history and provenance.
- Coordinator availability.

## Adversaries

- A malicious volunteer worker returning fabricated results.
- Several colluding workers.
- An attacker stealing an administrator key.
- A coordinator operator abusing access.
- A user submitting a harmful or privacy-invasive job.
- A denial-of-service attacker creating excessive jobs or leases.

## Existing alpha mitigations

- Worker tokens are stored as SHA-256 hashes, not plaintext.
- Work leases expire.
- Separate worker identities are required for replicated results by default.
- Accepted results require matching canonical hashes.
- Administrative writes require an admin key.
- Audit events record major writes and worker actions.

## Known gaps

- No fine-grained roles or organization boundaries.
- No rate limiting.
- No cryptographic signature on task definitions.
- No secure execution sandbox.
- No remote attestation.
- No Sybil resistance beyond controlled worker registration.
- No immutable or externally witnessed audit log.
- No privacy-budget accounting or differential privacy.
- No content safety review for submitted workloads.

## Recommended pilot rules

1. Permit only an allowlist of task implementations.
2. Permit only public or synthetic inputs.
3. Manually approve workers.
4. Place the coordinator behind TLS and an identity-aware proxy.
5. Monitor task volume and unusual result disagreements.
6. Keep a human appeal and correction process for operational uses.
```

### `examples/bootstrap_demo.py`

```python
import os

import httpx

SERVER = os.getenv("OIG_SERVER", "http://localhost:8000").rstrip("/")
ADMIN_KEY = os.getenv("OIG_ADMIN_KEY", "change-me-now")
HEADERS = {"X-Admin-Key": ADMIN_KEY}


def post(path: str, payload: dict) -> dict:
    response = httpx.post(f"{SERVER}{path}", headers=HEADERS, json=payload, timeout=30)
    response.raise_for_status()
    return response.json()


def main() -> None:
    organization = post(
        "/v1/object-types",
        {
            "name": "Organization",
            "description": "A community organization or institution.",
            "schema_json": {
                "type": "object",
                "properties": {"name": {"type": "string"}, "city": {"type": "string"}},
                "required": ["name"],
            },
        },
    )
    entity = post(
        "/v1/entities",
        {
            "object_type_id": organization["id"],
            "external_id": "demo-library",
            "properties": {"name": "Demo Public Library", "city": "Gloucester"},
        },
    )

    workers = []
    for name in ("demo-worker-a", "demo-worker-b"):
        workers.append(
            post(
                "/v1/workers/register",
                {"name": name, "capabilities": ["word_count", "sum_numbers", "hash_text", "json_stats"]},
            )
        )

    job = post(
        "/v1/jobs",
        {
            "task_type": "word_count",
            "input_payload": {
                "text": "Public knowledge should glow brighter when more hands carry the lantern."
            },
            "replicas_required": 2,
            "max_replicas": 3,
        },
    )

    print("Created entity:", entity["id"])
    print("Created job:", job["id"])
    for worker in workers:
        print(f"{worker['name']} token: {worker['token']}")


if __name__ == "__main__":
    main()
```

### `pyproject.toml`

```toml
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "open-intelligence-grid"
version = "0.1.0"
description = "An open-source ontology and volunteer-computing coordination platform."
readme = "README.md"
requires-python = ">=3.11"
license = "MIT"
authors = [{ name = "Open Intelligence Grid contributors" }]
dependencies = [
  "fastapi>=0.115,<1.0",
  "uvicorn[standard]>=0.30,<1.0",
  "sqlalchemy>=2.0,<2.2",
  "pydantic>=2.8,<3.0",
  "pydantic-settings>=2.4,<3.0",
  "httpx>=0.27,<1.0",
  "psycopg[binary]>=3.2,<4.0",
]

[project.optional-dependencies]
dev = [
  "pytest>=8.0,<9.0",
  "pytest-cov>=5.0,<7.0",
  "ruff>=0.8,<1.0",
]

[project.scripts]
oig-server = "oig.main:run"
oig-worker = "oig.worker:main"

[tool.setuptools]
package-dir = {"" = "src"}

[tool.setuptools.packages.find]
where = ["src"]

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-q"

[tool.ruff]
line-length = 100
target-version = "py311"
```

### `src/oig/__init__.py`

```python
"""Open Intelligence Grid core package."""

__version__ = "0.1.0"
```

### `src/oig/api.py`

```python
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, selectinload

from .db import get_db
from .models import AuditEvent, Entity, Job, ObjectType, Relationship, Worker, WorkUnit
from .schemas import (
    AuditRead,
    EntityCreate,
    EntityRead,
    GraphRead,
    JobCreate,
    JobRead,
    ObjectTypeCreate,
    ObjectTypeRead,
    RelationshipCreate,
    RelationshipRead,
    WorkerRead,
    WorkerRegister,
    WorkerRegistered,
    WorkClaim,
    WorkFailure,
    WorkSubmit,
)
from .security import hash_token, issue_token, require_admin, require_worker
from .services import audit, create_job, fail_work, graph_for_entity, lease_work, submit_result
from .tasks import TASKS

router = APIRouter(prefix="/v1")
Admin = Annotated[str, Depends(require_admin)]
DB = Annotated[Session, Depends(get_db)]
AuthenticatedWorker = Annotated[Worker, Depends(require_worker)]


@router.get("/stats")
def stats(db: DB) -> dict:
    job_rows = db.execute(select(Job.status, func.count(Job.id)).group_by(Job.status)).all()
    return {
        "object_types": db.scalar(select(func.count(ObjectType.id))) or 0,
        "entities": db.scalar(select(func.count(Entity.id))) or 0,
        "relationships": db.scalar(select(func.count(Relationship.id))) or 0,
        "workers": db.scalar(select(func.count(Worker.id))) or 0,
        "jobs": dict(job_rows),
        "supported_tasks": sorted(TASKS),
    }


@router.post("/object-types", response_model=ObjectTypeRead, status_code=status.HTTP_201_CREATED)
def create_object_type(payload: ObjectTypeCreate, db: DB, _admin: Admin) -> ObjectType:
    item = ObjectType(**payload.model_dump(by_alias=True))
    db.add(item)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="object_type.created",
        target_type="object_type",
        target_id=item.id,
        details={"name": item.name},
    )
    try:
        db.commit()
    except IntegrityError as exc:
        db.rollback()
        raise HTTPException(status_code=409, detail="Object type name already exists") from exc
    db.refresh(item)
    return item


@router.get("/object-types", response_model=list[ObjectTypeRead])
def list_object_types(db: DB, limit: int = Query(default=100, ge=1, le=500)) -> list[ObjectType]:
    return list(db.scalars(select(ObjectType).order_by(ObjectType.name).limit(limit)).all())


@router.post("/entities", response_model=EntityRead, status_code=status.HTTP_201_CREATED)
def create_entity(payload: EntityCreate, db: DB, _admin: Admin) -> Entity:
    if db.get(ObjectType, payload.object_type_id) is None:
        raise HTTPException(status_code=404, detail="Object type not found")
    entity = Entity(**payload.model_dump())
    db.add(entity)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="entity.created",
        target_type="entity",
        target_id=entity.id,
        details={"object_type_id": entity.object_type_id},
    )
    try:
        db.commit()
    except IntegrityError as exc:
        db.rollback()
        raise HTTPException(status_code=409, detail="Duplicate external ID for object type") from exc
    db.refresh(entity)
    return entity


@router.get("/entities", response_model=list[EntityRead])
def list_entities(
    db: DB,
    object_type_id: str | None = None,
    limit: int = Query(default=100, ge=1, le=500),
) -> list[Entity]:
    statement = select(Entity).order_by(Entity.created_at.desc()).limit(limit)
    if object_type_id:
        statement = statement.where(Entity.object_type_id == object_type_id)
    return list(db.scalars(statement).all())


@router.get("/entities/{entity_id}", response_model=EntityRead)
def get_entity(entity_id: str, db: DB) -> Entity:
    entity = db.get(Entity, entity_id)
    if entity is None:
        raise HTTPException(status_code=404, detail="Entity not found")
    return entity


@router.post(
    "/relationships", response_model=RelationshipRead, status_code=status.HTTP_201_CREATED
)
def create_relationship(payload: RelationshipCreate, db: DB, _admin: Admin) -> Relationship:
    if db.get(Entity, payload.source_entity_id) is None:
        raise HTTPException(status_code=404, detail="Source entity not found")
    if db.get(Entity, payload.target_entity_id) is None:
        raise HTTPException(status_code=404, detail="Target entity not found")
    relationship = Relationship(**payload.model_dump())
    db.add(relationship)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="relationship.created",
        target_type="relationship",
        target_id=relationship.id,
        details={"relationship_type": relationship.relationship_type},
    )
    db.commit()
    db.refresh(relationship)
    return relationship


@router.get("/relationships", response_model=list[RelationshipRead])
def list_relationships(db: DB, limit: int = Query(default=100, ge=1, le=500)) -> list[Relationship]:
    return list(
        db.scalars(select(Relationship).order_by(Relationship.created_at.desc()).limit(limit)).all()
    )


@router.get("/entities/{entity_id}/graph", response_model=GraphRead)
def entity_graph(entity_id: str, db: DB) -> GraphRead:
    graph = graph_for_entity(db, entity_id)
    if graph is None:
        raise HTTPException(status_code=404, detail="Entity not found")
    return graph


@router.post("/workers/register", response_model=WorkerRegistered, status_code=status.HTTP_201_CREATED)
def register_worker(payload: WorkerRegister, db: DB, _admin: Admin) -> WorkerRegistered:
    token = issue_token()
    worker = Worker(
        name=payload.name,
        capabilities=payload.capabilities,
        token_hash=hash_token(token),
    )
    db.add(worker)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="worker.registered",
        target_type="worker",
        target_id=worker.id,
        details={"name": worker.name, "capabilities": worker.capabilities},
    )
    db.commit()
    db.refresh(worker)
    return WorkerRegistered(
        id=worker.id,
        name=worker.name,
        capabilities=worker.capabilities,
        token=token,
    )


@router.get("/workers", response_model=list[WorkerRead])
def list_workers(db: DB, _admin: Admin) -> list[Worker]:
    return list(db.scalars(select(Worker).order_by(Worker.created_at.desc())).all())


@router.post("/jobs", response_model=JobRead, status_code=status.HTTP_201_CREATED)
def submit_job(payload: JobCreate, db: DB, _admin: Admin) -> Job:
    if payload.task_type not in TASKS:
        raise HTTPException(
            status_code=400,
            detail=f"Unsupported task type. Supported: {', '.join(sorted(TASKS))}",
        )
    return create_job(db, **payload.model_dump())


@router.get("/jobs", response_model=list[JobRead])
def list_jobs(
    db: DB,
    _admin: Admin,
    limit: int = Query(default=100, ge=1, le=500),
) -> list[Job]:
    return list(
        db.scalars(
            select(Job)
            .options(selectinload(Job.work_units))
            .order_by(Job.created_at.desc())
            .limit(limit)
        ).all()
    )


@router.get("/jobs/{job_id}", response_model=JobRead)
def get_job(job_id: str, db: DB, _admin: Admin) -> Job:
    job = db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job_id)
    )
    if job is None:
        raise HTTPException(status_code=404, detail="Job not found")
    return job


@router.post("/work/claim", response_model=WorkClaim | None)
def claim_work(db: DB, worker: AuthenticatedWorker, response: Response) -> WorkClaim | None:
    unit = lease_work(db, worker)
    if unit is None:
        response.status_code = status.HTTP_204_NO_CONTENT
        return None
    return WorkClaim(
        work_unit_id=unit.id,
        lease_token=unit.lease_token or "",
        lease_expires_at=unit.lease_expires_at,
        job_id=unit.job.id,
        task_type=unit.job.task_type,
        input_payload=unit.job.input_payload,
    )


@router.post("/work/{work_unit_id}/submit", response_model=JobRead)
def submit_work_result(
    work_unit_id: str,
    payload: WorkSubmit,
    db: DB,
    worker: AuthenticatedWorker,
) -> Job:
    unit = db.get(WorkUnit, work_unit_id)
    if unit is None:
        raise HTTPException(status_code=404, detail="Work unit not found")
    try:
        return submit_result(
            db,
            unit=unit,
            worker=worker,
            lease_token=payload.lease_token,
            result_payload=payload.result_payload,
        )
    except ValueError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@router.post("/work/{work_unit_id}/fail", response_model=JobRead)
def report_work_failure(
    work_unit_id: str,
    payload: WorkFailure,
    db: DB,
    worker: AuthenticatedWorker,
) -> Job:
    unit = db.get(WorkUnit, work_unit_id)
    if unit is None:
        raise HTTPException(status_code=404, detail="Work unit not found")
    try:
        return fail_work(
            db,
            unit=unit,
            worker=worker,
            lease_token=payload.lease_token,
            error=payload.error,
        )
    except ValueError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@router.get("/audit", response_model=list[AuditRead])
def list_audit_events(
    db: DB,
    _admin: Admin,
    limit: int = Query(default=100, ge=1, le=500),
) -> list[AuditEvent]:
    return list(
        db.scalars(select(AuditEvent).order_by(AuditEvent.created_at.desc()).limit(limit)).all()
    )
```

### `src/oig/config.py`

```python
from functools import lru_cache

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="OIG_",
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    app_name: str = "Open Intelligence Grid"
    environment: str = "development"
    database_url: str = "sqlite:///./oig.db"
    admin_key: str = "change-me-now"
    lease_seconds: int = 300
    allow_same_worker_replicas: bool = False
    cors_origins: list[str] = Field(default_factory=list)


@lru_cache
def get_settings() -> Settings:
    return Settings()
```

### `src/oig/db.py`

```python
from collections.abc import Generator

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

from .config import get_settings


class Base(DeclarativeBase):
    pass


_settings = get_settings()
_connect_args = {"check_same_thread": False} if _settings.database_url.startswith("sqlite") else {}
engine = create_engine(_settings.database_url, connect_args=_connect_args, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)


def init_db() -> None:
    from . import models  # noqa: F401

    Base.metadata.create_all(bind=engine)


def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
```

### `src/oig/main.py`

```python
from contextlib import asynccontextmanager

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse

from .api import router
from .config import get_settings
from .db import init_db


DASHBOARD = """
<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Open Intelligence Grid</title>
  <style>
    :root { color-scheme: dark; font-family: ui-monospace, SFMono-Regular, Menlo, monospace; }
    body { margin: 0; background: #101419; color: #e8f0e8; }
    main { max-width: 960px; margin: 0 auto; padding: 48px 24px; }
    h1 { font-size: clamp(2rem, 6vw, 4.5rem); margin: 0; letter-spacing: -0.06em; }
    .subtitle { color: #9db0a2; max-width: 720px; line-height: 1.6; }
    .grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(150px, 1fr)); gap: 12px; margin: 32px 0; }
    .card { border: 1px solid #334139; border-radius: 12px; padding: 18px; background: #151c19; }
    .number { font-size: 2rem; font-weight: 700; }
    a { color: #a9ffbe; }
    pre { white-space: pre-wrap; word-break: break-word; background: #0b0e0c; padding: 16px; border-radius: 12px; }
    .pulse { display: inline-block; width: 10px; height: 10px; border-radius: 50%; background: #a9ffbe; box-shadow: 0 0 16px #a9ffbe; }
  </style>
</head>
<body>
<main>
  <div><span class="pulse"></span> alpha coordinator online</div>
  <h1>Open Intelligence Grid</h1>
  <p class="subtitle">An inspectable ontology and distributed-computing foundation. Small enough to understand. Large enough to begin.</p>
  <div class="grid" id="stats"><div class="card">Loading statistics...</div></div>
  <p><a href="/docs">Interactive API documentation</a> · <a href="/health">Health check</a></p>
  <h2>Supported worker tasks</h2>
  <pre id="tasks">Loading...</pre>
</main>
<script>
fetch('/v1/stats').then(r => r.json()).then(data => {
  const entries = [
    ['Object types', data.object_types], ['Entities', data.entities],
    ['Relationships', data.relationships], ['Workers', data.workers],
    ['Jobs', Object.values(data.jobs).reduce((a,b) => a+b, 0)]
  ];
  document.getElementById('stats').innerHTML = entries.map(([k,v]) => `<div class="card"><div class="number">${v}</div><div>${k}</div></div>`).join('');
  document.getElementById('tasks').textContent = data.supported_tasks.join('\n');
}).catch(error => {
  document.getElementById('stats').innerHTML = `<div class="card">Could not load stats: ${error}</div>`;
});
</script>
</body>
</html>
"""


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_db()
    yield


settings = get_settings()
app = FastAPI(
    title=settings.app_name,
    version="0.1.0",
    description="Open ontology and distributed volunteer-computing coordinator.",
    lifespan=lifespan,
)
if settings.cors_origins:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=False,
        allow_methods=["GET", "POST", "PATCH", "DELETE"],
        allow_headers=["Authorization", "Content-Type", "X-Admin-Key"],
    )
app.include_router(router)


@app.get("/", response_class=HTMLResponse, include_in_schema=False)
def dashboard() -> str:
    return DASHBOARD


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok", "version": "0.1.0", "environment": settings.environment}


def run() -> None:
    uvicorn.run("oig.main:app", host="0.0.0.0", port=8000, reload=False)


if __name__ == "__main__":
    run()
```

### `src/oig/models.py`

```python
from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, JSON, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship

from .db import Base


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


def uuid_string() -> str:
    return str(uuid.uuid4())


class ObjectType(Base):
    __tablename__ = "object_types"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    name: Mapped[str] = mapped_column(String(120), unique=True, index=True)
    description: Mapped[str] = mapped_column(Text, default="")
    schema_json: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)

    entities: Mapped[list[Entity]] = relationship(back_populates="object_type")


class Entity(Base):
    __tablename__ = "entities"
    __table_args__ = (UniqueConstraint("object_type_id", "external_id", name="uq_entity_external"),)

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    object_type_id: Mapped[str] = mapped_column(ForeignKey("object_types.id"), index=True)
    external_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
    properties: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow)

    object_type: Mapped[ObjectType] = relationship(back_populates="entities")


class Relationship(Base):
    __tablename__ = "relationships"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    relationship_type: Mapped[str] = mapped_column(String(120), index=True)
    source_entity_id: Mapped[str] = mapped_column(ForeignKey("entities.id"), index=True)
    target_entity_id: Mapped[str] = mapped_column(ForeignKey("entities.id"), index=True)
    properties: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)

    source: Mapped[Entity] = relationship(foreign_keys=[source_entity_id])
    target: Mapped[Entity] = relationship(foreign_keys=[target_entity_id])


class Worker(Base):
    __tablename__ = "workers"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    name: Mapped[str] = mapped_column(String(160))
    token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True)
    capabilities: Mapped[list[str]] = mapped_column(JSON, default=list)
    trust_score: Mapped[float] = mapped_column(Float, default=1.0)
    active: Mapped[bool] = mapped_column(Boolean, default=True)
    last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)


class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    task_type: Mapped[str] = mapped_column(String(120), index=True)
    input_payload: Mapped[dict[str, Any]] = mapped_column(JSON)
    status: Mapped[str] = mapped_column(String(30), default="queued", index=True)
    replicas_required: Mapped[int] = mapped_column(Integer, default=2)
    max_replicas: Mapped[int] = mapped_column(Integer, default=3)
    result_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
    result_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow)

    work_units: Mapped[list[WorkUnit]] = relationship(
        back_populates="job", cascade="all, delete-orphan", order_by="WorkUnit.created_at"
    )


class WorkUnit(Base):
    __tablename__ = "work_units"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    job_id: Mapped[str] = mapped_column(ForeignKey("jobs.id"), index=True)
    replica_index: Mapped[int] = mapped_column(Integer)
    status: Mapped[str] = mapped_column(String(30), default="queued", index=True)
    worker_id: Mapped[str | None] = mapped_column(ForeignKey("workers.id"), nullable=True, index=True)
    lease_token: Mapped[str | None] = mapped_column(String(64), nullable=True)
    lease_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    result_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
    result_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    error: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    job: Mapped[Job] = relationship(back_populates="work_units")
    worker: Mapped[Worker | None] = relationship()


class AuditEvent(Base):
    __tablename__ = "audit_events"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    actor_type: Mapped[str] = mapped_column(String(40))
    actor_id: Mapped[str] = mapped_column(String(160))
    action: Mapped[str] = mapped_column(String(160), index=True)
    target_type: Mapped[str] = mapped_column(String(80))
    target_id: Mapped[str] = mapped_column(String(160))
    details: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, index=True)
```

### `src/oig/schemas.py`

```python
from datetime import datetime
from typing import Any

from pydantic import BaseModel, ConfigDict, Field, model_validator


class ORMModel(BaseModel):
    model_config = ConfigDict(from_attributes=True)


class ObjectTypeCreate(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    name: str = Field(min_length=1, max_length=120, pattern=r"^[A-Za-z][A-Za-z0-9_-]*$")
    description: str = ""
    schema_definition: dict[str, Any] = Field(default_factory=dict, alias="schema_json")


class ObjectTypeRead(ORMModel):
    id: str
    name: str
    description: str
    schema_definition: dict[str, Any] = Field(
        validation_alias="schema_json", serialization_alias="schema_json"
    )
    created_at: datetime


class EntityCreate(BaseModel):
    object_type_id: str
    external_id: str | None = None
    properties: dict[str, Any] = Field(default_factory=dict)


class EntityRead(ORMModel):
    id: str
    object_type_id: str
    external_id: str | None
    properties: dict[str, Any]
    created_at: datetime
    updated_at: datetime


class RelationshipCreate(BaseModel):
    relationship_type: str = Field(min_length=1, max_length=120)
    source_entity_id: str
    target_entity_id: str
    properties: dict[str, Any] = Field(default_factory=dict)


class RelationshipRead(ORMModel):
    id: str
    relationship_type: str
    source_entity_id: str
    target_entity_id: str
    properties: dict[str, Any]
    created_at: datetime


class GraphRead(BaseModel):
    center: EntityRead
    entities: list[EntityRead]
    relationships: list[RelationshipRead]


class WorkerRegister(BaseModel):
    name: str = Field(min_length=1, max_length=160)
    capabilities: list[str] = Field(default_factory=list)


class WorkerRegistered(BaseModel):
    id: str
    name: str
    capabilities: list[str]
    token: str


class WorkerRead(ORMModel):
    id: str
    name: str
    capabilities: list[str]
    trust_score: float
    active: bool
    last_seen_at: datetime | None
    created_at: datetime


class JobCreate(BaseModel):
    task_type: str = Field(min_length=1, max_length=120)
    input_payload: dict[str, Any]
    replicas_required: int = Field(default=2, ge=1, le=10)
    max_replicas: int = Field(default=3, ge=1, le=20)

    @model_validator(mode="after")
    def validate_replica_counts(self) -> "JobCreate":
        if self.max_replicas < self.replicas_required:
            raise ValueError("max_replicas must be greater than or equal to replicas_required")
        return self


class WorkUnitRead(ORMModel):
    id: str
    job_id: str
    replica_index: int
    status: str
    worker_id: str | None
    lease_expires_at: datetime | None
    result_hash: str | None
    error: str | None
    created_at: datetime
    completed_at: datetime | None


class JobRead(ORMModel):
    id: str
    task_type: str
    input_payload: dict[str, Any]
    status: str
    replicas_required: int
    max_replicas: int
    result_payload: dict[str, Any] | None
    result_hash: str | None
    created_at: datetime
    updated_at: datetime
    work_units: list[WorkUnitRead] = Field(default_factory=list)


class WorkClaim(BaseModel):
    work_unit_id: str
    lease_token: str
    lease_expires_at: datetime
    job_id: str
    task_type: str
    input_payload: dict[str, Any]


class WorkSubmit(BaseModel):
    lease_token: str
    result_payload: dict[str, Any]


class WorkFailure(BaseModel):
    lease_token: str
    error: str = Field(min_length=1, max_length=4000)


class AuditRead(ORMModel):
    id: int
    actor_type: str
    actor_id: str
    action: str
    target_type: str
    target_id: str
    details: dict[str, Any]
    created_at: datetime
```

### `src/oig/security.py`

```python
import hashlib
import secrets

from fastapi import Depends, Header, HTTPException, status
from sqlalchemy import select
from sqlalchemy.orm import Session

from .config import get_settings
from .db import get_db
from .models import Worker, utcnow


def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def issue_token() -> str:
    return secrets.token_urlsafe(32)


def require_admin(x_admin_key: str = Header(default="")) -> str:
    settings = get_settings()
    if not x_admin_key or not secrets.compare_digest(x_admin_key, settings.admin_key):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid admin key")
    return "admin"


def require_worker(
    authorization: str = Header(default=""), db: Session = Depends(get_db)
) -> Worker:
    scheme, _, token = authorization.partition(" ")
    if scheme.lower() != "bearer" or not token:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Bearer token required")

    worker = db.scalar(select(Worker).where(Worker.token_hash == hash_token(token)))
    if worker is None or not worker.active:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid worker token")

    worker.last_seen_at = utcnow()
    db.commit()
    db.refresh(worker)
    return worker
```

### `src/oig/services.py`

```python
from __future__ import annotations

import hashlib
import json
import secrets
from collections import Counter
from datetime import datetime, timedelta, timezone
from typing import Any

from sqlalchemy import or_, select
from sqlalchemy.orm import Session, selectinload

from .config import get_settings
from .models import AuditEvent, Entity, Job, Relationship, Worker, WorkUnit, utcnow
from .schemas import GraphRead


def _as_utc(value: datetime) -> datetime:
    """Normalize timestamps returned by SQLite, which may drop timezone metadata."""
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)


def canonical_hash(payload: dict[str, Any]) -> str:
    encoded = json.dumps(
        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def audit(
    db: Session,
    *,
    actor_type: str,
    actor_id: str,
    action: str,
    target_type: str,
    target_id: str,
    details: dict[str, Any] | None = None,
) -> None:
    db.add(
        AuditEvent(
            actor_type=actor_type,
            actor_id=actor_id,
            action=action,
            target_type=target_type,
            target_id=target_id,
            details=details or {},
        )
    )


def create_job(db: Session, *, task_type: str, input_payload: dict[str, Any], replicas_required: int, max_replicas: int) -> Job:
    job = Job(
        task_type=task_type,
        input_payload=input_payload,
        replicas_required=replicas_required,
        max_replicas=max_replicas,
    )
    db.add(job)
    db.flush()
    for replica_index in range(replicas_required):
        db.add(WorkUnit(job_id=job.id, replica_index=replica_index))
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="job.created",
        target_type="job",
        target_id=job.id,
        details={"task_type": task_type, "replicas_required": replicas_required},
    )
    db.commit()
    return db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job.id)
    )


def _release_expired_leases(db: Session) -> None:
    now = utcnow()
    expired = db.scalars(
        select(WorkUnit).where(
            WorkUnit.status == "leased",
            WorkUnit.lease_expires_at.is_not(None),
            WorkUnit.lease_expires_at < now,
        )
    ).all()
    for unit in expired:
        unit.status = "queued"
        unit.worker_id = None
        unit.lease_token = None
        unit.lease_expires_at = None
    if expired:
        db.commit()


def lease_work(db: Session, worker: Worker) -> WorkUnit | None:
    settings = get_settings()
    _release_expired_leases(db)

    candidates = db.scalars(
        select(WorkUnit)
        .options(selectinload(WorkUnit.job))
        .where(WorkUnit.status == "queued")
        .order_by(WorkUnit.created_at)
    ).all()

    for unit in candidates:
        job = unit.job
        if job.status not in {"queued", "running"}:
            continue
        if worker.capabilities and job.task_type not in worker.capabilities:
            continue
        if not settings.allow_same_worker_replicas:
            already_worked = db.scalar(
                select(WorkUnit.id).where(
                    WorkUnit.job_id == job.id,
                    WorkUnit.worker_id == worker.id,
                )
            )
            if already_worked is not None:
                continue

        unit.status = "leased"
        unit.worker_id = worker.id
        unit.lease_token = secrets.token_urlsafe(24)
        unit.lease_expires_at = utcnow() + timedelta(seconds=settings.lease_seconds)
        job.status = "running"
        audit(
            db,
            actor_type="worker",
            actor_id=worker.id,
            action="work.leased",
            target_type="work_unit",
            target_id=unit.id,
            details={"job_id": job.id},
        )
        db.commit()
        db.refresh(unit)
        return unit
    return None


def _evaluate_job(db: Session, job: Job) -> None:
    completed = [unit for unit in job.work_units if unit.status == "completed" and unit.result_hash]
    counts = Counter(unit.result_hash for unit in completed)

    if counts:
        winning_hash, winning_count = counts.most_common(1)[0]
        if winning_count >= job.replicas_required:
            winning_unit = next(unit for unit in completed if unit.result_hash == winning_hash)
            job.status = "completed"
            job.result_hash = winning_hash
            job.result_payload = winning_unit.result_payload
            for unit in job.work_units:
                if unit.status == "queued":
                    unit.status = "cancelled"
            audit(
                db,
                actor_type="system",
                actor_id="consensus",
                action="job.completed",
                target_type="job",
                target_id=job.id,
                details={"result_hash": winning_hash, "matching_replicas": winning_count},
            )
            return

    active_or_waiting = [unit for unit in job.work_units if unit.status in {"queued", "leased"}]
    if active_or_waiting:
        return

    if len(job.work_units) < job.max_replicas:
        next_index = max((unit.replica_index for unit in job.work_units), default=-1) + 1
        job.work_units.append(WorkUnit(job_id=job.id, replica_index=next_index))
        job.status = "running"
        audit(
            db,
            actor_type="system",
            actor_id="consensus",
            action="work.tie_breaker_created",
            target_type="job",
            target_id=job.id,
            details={"replica_index": next_index},
        )
    else:
        job.status = "disputed"
        audit(
            db,
            actor_type="system",
            actor_id="consensus",
            action="job.disputed",
            target_type="job",
            target_id=job.id,
            details={"hash_counts": dict(counts)},
        )


def submit_result(
    db: Session,
    *,
    unit: WorkUnit,
    worker: Worker,
    lease_token: str,
    result_payload: dict[str, Any],
) -> Job:
    if unit.status != "leased" or unit.worker_id != worker.id:
        raise ValueError("Work unit is not leased to this worker")
    if not unit.lease_token or not secrets.compare_digest(unit.lease_token, lease_token):
        raise ValueError("Invalid lease token")
    if unit.lease_expires_at and _as_utc(unit.lease_expires_at) < utcnow():
        raise ValueError("Lease expired")

    unit.status = "completed"
    unit.result_payload = result_payload
    unit.result_hash = canonical_hash(result_payload)
    unit.completed_at = utcnow()
    unit.lease_token = None
    unit.lease_expires_at = None
    worker.trust_score = min(worker.trust_score + 0.01, 10.0)
    audit(
        db,
        actor_type="worker",
        actor_id=worker.id,
        action="work.completed",
        target_type="work_unit",
        target_id=unit.id,
        details={"result_hash": unit.result_hash},
    )

    db.flush()
    job = db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == unit.job_id)
    )
    if job is None:
        raise ValueError("Parent job not found")
    _evaluate_job(db, job)
    db.commit()
    return db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job.id)
    )


def fail_work(
    db: Session, *, unit: WorkUnit, worker: Worker, lease_token: str, error: str
) -> Job:
    if unit.status != "leased" or unit.worker_id != worker.id:
        raise ValueError("Work unit is not leased to this worker")
    if not unit.lease_token or not secrets.compare_digest(unit.lease_token, lease_token):
        raise ValueError("Invalid lease token")

    unit.status = "failed"
    unit.error = error
    unit.completed_at = utcnow()
    unit.lease_token = None
    unit.lease_expires_at = None
    worker.trust_score = max(worker.trust_score - 0.02, 0.0)

    db.flush()
    job = db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == unit.job_id)
    )
    if job is None:
        raise ValueError("Parent job not found")
    _evaluate_job(db, job)
    db.commit()
    return db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job.id)
    )


def graph_for_entity(db: Session, entity_id: str) -> GraphRead | None:
    center = db.get(Entity, entity_id)
    if center is None:
        return None
    relationships = db.scalars(
        select(Relationship).where(
            or_(
                Relationship.source_entity_id == entity_id,
                Relationship.target_entity_id == entity_id,
            )
        )
    ).all()
    neighbor_ids = {
        rel.target_entity_id if rel.source_entity_id == entity_id else rel.source_entity_id
        for rel in relationships
    }
    neighbors = db.scalars(select(Entity).where(Entity.id.in_(neighbor_ids))).all() if neighbor_ids else []
    return GraphRead(center=center, entities=[center, *neighbors], relationships=relationships)
```

### `src/oig/tasks.py`

```python
import hashlib
import json
from collections.abc import Callable
from typing import Any


class TaskError(ValueError):
    pass


def word_count(payload: dict[str, Any]) -> dict[str, Any]:
    text = payload.get("text")
    if not isinstance(text, str):
        raise TaskError("word_count requires a string field named 'text'")
    return {
        "words": len(text.split()),
        "characters": len(text),
        "lines": len(text.splitlines()) if text else 0,
    }


def sum_numbers(payload: dict[str, Any]) -> dict[str, Any]:
    numbers = payload.get("numbers")
    if not isinstance(numbers, list) or not all(
        isinstance(value, int) and not isinstance(value, bool) for value in numbers
    ):
        raise TaskError("sum_numbers requires an integer list field named 'numbers'")
    return {"sum": sum(numbers), "count": len(numbers)}


def hash_text(payload: dict[str, Any]) -> dict[str, Any]:
    text = payload.get("text")
    algorithm = payload.get("algorithm", "sha256")
    if not isinstance(text, str):
        raise TaskError("hash_text requires a string field named 'text'")
    if algorithm not in {"sha256", "sha512", "blake2b"}:
        raise TaskError("algorithm must be sha256, sha512, or blake2b")
    digest = hashlib.new(algorithm, text.encode("utf-8")).hexdigest()
    return {"algorithm": algorithm, "digest": digest}


def json_stats(payload: dict[str, Any]) -> dict[str, Any]:
    def walk(value: Any) -> tuple[int, int, int, int]:
        if isinstance(value, dict):
            child = [walk(item) for item in value.values()]
            return (
                1 + sum(item[0] for item in child),
                sum(item[1] for item in child),
                sum(item[2] for item in child),
                sum(item[3] for item in child),
            )
        if isinstance(value, list):
            child = [walk(item) for item in value]
            return (
                sum(item[0] for item in child),
                1 + sum(item[1] for item in child),
                sum(item[2] for item in child),
                sum(item[3] for item in child),
            )
        if isinstance(value, (str, int, float, bool)) or value is None:
            return (0, 0, 1, len(json.dumps(value, ensure_ascii=False)))
        raise TaskError("payload contains an unsupported JSON value")

    objects, arrays, scalars, scalar_bytes = walk(payload)
    return {
        "objects": objects,
        "arrays": arrays,
        "scalars": scalars,
        "serialized_scalar_characters": scalar_bytes,
    }


TASKS: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = {
    "word_count": word_count,
    "sum_numbers": sum_numbers,
    "hash_text": hash_text,
    "json_stats": json_stats,
}


def run_task(task_type: str, payload: dict[str, Any]) -> dict[str, Any]:
    task = TASKS.get(task_type)
    if task is None:
        raise TaskError(f"Unsupported task type: {task_type}")
    return task(payload)
```

### `src/oig/worker.py`

```python
import argparse
import sys
import time

import httpx

from .tasks import TaskError, run_task


def worker_headers(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def process_once(server: str, token: str, timeout: float = 30.0) -> bool:
    server = server.rstrip("/")
    with httpx.Client(timeout=timeout, headers=worker_headers(token)) as client:
        claim = client.post(f"{server}/v1/work/claim")
        if claim.status_code == 204:
            return False
        claim.raise_for_status()
        work = claim.json()

        try:
            result = run_task(work["task_type"], work["input_payload"])
        except (TaskError, Exception) as exc:
            failure = client.post(
                f"{server}/v1/work/{work['work_unit_id']}/fail",
                json={"lease_token": work["lease_token"], "error": str(exc)},
            )
            failure.raise_for_status()
            print(f"Failed work unit {work['work_unit_id']}: {exc}", file=sys.stderr)
            return True

        response = client.post(
            f"{server}/v1/work/{work['work_unit_id']}/submit",
            json={"lease_token": work["lease_token"], "result_payload": result},
        )
        response.raise_for_status()
        job = response.json()
        print(
            f"Completed work unit {work['work_unit_id']} for job {work['job_id']} "
            f"(job status: {job['status']})"
        )
        return True


def main() -> None:
    parser = argparse.ArgumentParser(description="Open Intelligence Grid volunteer worker")
    parser.add_argument("--server", default="http://localhost:8000")
    parser.add_argument("--token", required=True, help="Worker token returned during registration")
    parser.add_argument("--once", action="store_true", help="Claim at most one work unit and exit")
    parser.add_argument("--poll-seconds", type=float, default=10.0)
    args = parser.parse_args()

    while True:
        try:
            worked = process_once(args.server, args.token)
        except httpx.HTTPError as exc:
            print(f"Coordinator request failed: {exc}", file=sys.stderr)
            worked = False

        if args.once:
            return
        if not worked:
            time.sleep(max(args.poll_seconds, 1.0))


if __name__ == "__main__":
    main()
```

### `tests/conftest.py`

```python
from pathlib import Path

import pytest
from fastapi.testclient import TestClient


@pytest.fixture()
def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    database_path = tmp_path / "test.db"
    monkeypatch.setenv("OIG_DATABASE_URL", f"sqlite:///{database_path}")
    monkeypatch.setenv("OIG_ADMIN_KEY", "test-admin")
    monkeypatch.setenv("OIG_ALLOW_SAME_WORKER_REPLICAS", "false")

    import oig.config as config

    config.get_settings.cache_clear()

    import oig.db as db

    db.engine.dispose()
    connect_args = {"check_same_thread": False}
    db.engine = db.create_engine(
        f"sqlite:///{database_path}", connect_args=connect_args, pool_pre_ping=True
    )
    db.SessionLocal.configure(bind=db.engine)

    from oig.main import app

    with TestClient(app) as test_client:
        yield test_client
```

### `tests/test_health.py`

```python
def test_health(client):
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "ok"
```

### `tests/test_jobs.py`

```python
ADMIN = {"X-Admin-Key": "test-admin"}


def register_worker(client, name):
    response = client.post(
        "/v1/workers/register",
        headers=ADMIN,
        json={"name": name, "capabilities": ["sum_numbers"]},
    )
    assert response.status_code == 201
    return response.json()


def claim(client, token):
    response = client.post(
        "/v1/work/claim", headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 200
    return response.json()


def submit(client, token, work, result):
    return client.post(
        f"/v1/work/{work['work_unit_id']}/submit",
        headers={"Authorization": f"Bearer {token}"},
        json={"lease_token": work["lease_token"], "result_payload": result},
    )


def test_two_workers_reach_consensus(client):
    worker_a = register_worker(client, "a")
    worker_b = register_worker(client, "b")

    job = client.post(
        "/v1/jobs",
        headers=ADMIN,
        json={
            "task_type": "sum_numbers",
            "input_payload": {"numbers": [1, 2, 3]},
            "replicas_required": 2,
            "max_replicas": 3,
        },
    )
    assert job.status_code == 201

    work_a = claim(client, worker_a["token"])
    work_b = claim(client, worker_b["token"])

    result_a = submit(client, worker_a["token"], work_a, {"sum": 6, "count": 3})
    result_b = submit(client, worker_b["token"], work_b, {"count": 3, "sum": 6})
    assert result_a.status_code == 200
    assert result_b.status_code == 200
    assert result_b.json()["status"] == "completed"
    assert result_b.json()["result_payload"] == {"count": 3, "sum": 6}


def test_disagreement_creates_tie_breaker(client):
    workers = [register_worker(client, name) for name in ("a", "b", "c")]
    client.post(
        "/v1/jobs",
        headers=ADMIN,
        json={
            "task_type": "sum_numbers",
            "input_payload": {"numbers": [4, 5]},
            "replicas_required": 2,
            "max_replicas": 3,
        },
    ).json()

    first = claim(client, workers[0]["token"])
    second = claim(client, workers[1]["token"])
    submit(client, workers[0]["token"], first, {"sum": 9, "count": 2})
    disputed_so_far = submit(client, workers[1]["token"], second, {"sum": 99, "count": 2})
    assert disputed_so_far.status_code == 200
    assert len(disputed_so_far.json()["work_units"]) == 3

    tie_breaker = claim(client, workers[2]["token"])
    final = submit(client, workers[2]["token"], tie_breaker, {"count": 2, "sum": 9})
    assert final.status_code == 200
    assert final.json()["status"] == "completed"
    assert final.json()["result_payload"]["sum"] == 9
```

### `tests/test_ontology.py`

```python
ADMIN = {"X-Admin-Key": "test-admin"}


def test_create_entities_and_graph(client):
    org_type = client.post(
        "/v1/object-types",
        headers=ADMIN,
        json={"name": "Organization", "schema_json": {"type": "object"}},
    )
    assert org_type.status_code == 201

    program_type = client.post(
        "/v1/object-types",
        headers=ADMIN,
        json={"name": "Program", "schema_json": {"type": "object"}},
    )
    assert program_type.status_code == 201

    org = client.post(
        "/v1/entities",
        headers=ADMIN,
        json={"object_type_id": org_type.json()["id"], "properties": {"name": "Library"}},
    )
    program = client.post(
        "/v1/entities",
        headers=ADMIN,
        json={"object_type_id": program_type.json()["id"], "properties": {"name": "Robotics"}},
    )
    assert org.status_code == 201
    assert program.status_code == 201

    relation = client.post(
        "/v1/relationships",
        headers=ADMIN,
        json={
            "relationship_type": "OPERATES",
            "source_entity_id": org.json()["id"],
            "target_entity_id": program.json()["id"],
            "properties": {},
        },
    )
    assert relation.status_code == 201

    graph = client.get(f"/v1/entities/{org.json()['id']}/graph")
    assert graph.status_code == 200
    assert len(graph.json()["entities"]) == 2
    assert graph.json()["relationships"][0]["relationship_type"] == "OPERATES"
```

 

Here is the source code: 

OPEN INTELLIGENCE GRID v0.1.0 — COMPLETE SOURCE CODE APPENDIX

Repository tree:
.dockerignore
.env.example
.github/workflows/ci.yml
.gitignore
CHANGELOG.md
CODE_OF_CONDUCT.md
CONTRIBUTING.md
Dockerfile
GOVERNANCE.md
LICENSE
Makefile
README.md
SECURITY.md
compose.yaml
docs/API.md
docs/ARCHITECTURE.md
docs/GITHUB_PUBLISHING.md
docs/QUICKSTART.md
docs/RELEASE_VALIDATION.md
docs/ROADMAP.md
docs/THREAT_MODEL.md
examples/bootstrap_demo.py
pyproject.toml
src/oig/__init__.py
src/oig/api.py
src/oig/config.py
src/oig/db.py
src/oig/main.py
src/oig/models.py
src/oig/schemas.py
src/oig/security.py
src/oig/services.py
src/oig/tasks.py
src/oig/worker.py
tests/conftest.py
tests/test_health.py
tests/test_jobs.py
tests/test_ontology.py

================================================================================
FILE: .dockerignore
================================================================================
.git
.venv
__pycache__
.pytest_cache
*.db
.env

================================================================================
FILE: .env.example
================================================================================
OIG_APP_NAME=Open Intelligence Grid
OIG_ENVIRONMENT=development
OIG_DATABASE_URL=sqlite:///./oig.db
OIG_ADMIN_KEY=change-me-now
OIG_LEASE_SECONDS=300
OIG_ALLOW_SAME_WORKER_REPLICAS=false
OIG_CORS_ORIGINS=[]

================================================================================
FILE: .github/workflows/ci.yml
================================================================================
name: CI

on:
  push:
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.11", "3.12", "3.13"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
      - run: python -m pip install --upgrade pip
      - run: pip install -e '.[dev]'
      - run: ruff check src tests
      - run: pytest

================================================================================
FILE: .gitignore
================================================================================
.venv/
__pycache__/
*.py[cod]
.pytest_cache/
.ruff_cache/
.coverage
htmlcov/
.env
*.db
*.sqlite3
dist/
build/
*.egg-info/
.DS_Store

================================================================================
FILE: CHANGELOG.md
================================================================================
# Changelog

## 0.1.0 - 2026-07-01

### Added

- Ontology object types, entities, and relationships.
- One-hop graph traversal.
- Worker registration and bearer-token authentication.
- Database-backed work leasing.
- Replicated deterministic jobs with hash-based consensus.
- Tie-breaking work units for conflicting results.
- SQLite and PostgreSQL support.
- Docker Compose environment.
- Minimal browser dashboard.
- Tests and initial security documentation.

================================================================================
FILE: CODE_OF_CONDUCT.md
================================================================================
# Community code of conduct

## Our standard

Participants should help create a technically serious, curious, and humane community. Expected behavior includes:

- Critiquing ideas and code without degrading people.
- Welcoming contributors with different levels of experience.
- Documenting decisions that affect users or communities.
- Respecting privacy, consent, and the right to correct errors.
- Disclosing conflicts of interest.
- Avoiding harassment, threats, discrimination, and deliberate disruption.

## Public-interest responsibility

This project can become powerful. Contributors must not use project spaces to promote covert surveillance, harassment, discriminatory targeting, or collection of personal information without a lawful and ethical basis.

## Enforcement

Before public launch, maintainers should publish a dedicated conduct contact, a private reporting method, an appeal procedure, and a transparent range of enforcement actions.

================================================================================
FILE: CONTRIBUTING.md
================================================================================
# Contributing

Thank you for helping build an inspectable public-interest data platform.

## Development setup

```bash
python -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]'
pytest
ruff check src tests
```

## Contribution rules

1. Keep pull requests focused and documented.
2. Add tests for behavior changes.
3. Do not add surveillance-specific features without a public-interest review.
4. Do not add dependencies without explaining why the standard library or an existing dependency is insufficient.
5. Never commit secrets or real personal data.
6. Preserve backward compatibility or document a migration path.

## Commit suggestions

Use clear prefixes such as `feat:`, `fix:`, `docs:`, `test:`, and `security:`.

## Definition of done

A change should include implementation, tests, documentation, and a note in `CHANGELOG.md` when users will notice it.

================================================================================
FILE: Dockerfile
================================================================================
FROM python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

WORKDIR /app

COPY pyproject.toml README.md LICENSE ./
COPY src ./src
RUN pip install --no-cache-dir .

EXPOSE 8000
CMD ["uvicorn", "oig.main:app", "--host", "0.0.0.0", "--port", "8000"]

================================================================================
FILE: GOVERNANCE.md
================================================================================
# Governance draft

OIG should eventually be governed by a nonprofit or multistakeholder foundation rather than a single vendor.

## Proposed bodies

- **Maintainers:** accept code and manage releases.
- **Security team:** coordinates private vulnerability reports.
- **Public-interest council:** reviews high-risk capabilities and deployment patterns.
- **Data-governance working group:** develops consent, retention, provenance, and redress standards.
- **Community assembly:** proposes roadmap items and elects or confirms council members.

## High-risk change review

Features involving identity resolution, location tracking, biometric inference, predictive policing, mass scraping, or covert monitoring should require an explicit written review and documented safeguards before merging.

This is a starting constitution, not stone tablets descending from a venture-capital cloud.

================================================================================
FILE: LICENSE
================================================================================
MIT License

Copyright (c) 2026 Open Intelligence Grid contributors

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

================================================================================
FILE: Makefile
================================================================================
.PHONY: install dev test lint run docker-up docker-down

install:
python -m pip install -e .

dev:
python -m pip install -e '.[dev]'

test:
pytest

lint:
ruff check src tests

run:
oig-server

docker-up:
docker compose up --build

docker-down:
docker compose down

================================================================================
FILE: README.md
================================================================================
# Open Intelligence Grid

**Open Intelligence Grid (OIG)** is an early, working open-source foundation for a transparent data-and-operations platform with distributed volunteer computing.

It combines two ideas:

1. An **ontology layer** that represents real-world object types, entities, and relationships.
2. A **coordinator/worker network** that divides deterministic jobs into independently validated work units.

This is a usable alpha, not a production replacement for Palantir Foundry. It is intentionally small enough to understand, run, audit, and improve.

## What works in v0.1.0

- Define ontology object types with JSON Schema-like property descriptions.
- Create entities and relationships.
- Traverse a one-hop entity graph.
- Register authenticated volunteer workers.
- Submit distributed jobs with configurable replication.
- Lease work units to separate workers.
- Validate results by canonical-hash consensus.
- Add a tie-breaking replica when workers disagree.
- Record basic audit events.
- Run on SQLite for development or PostgreSQL with Docker Compose.
- Use the bundled deterministic worker tasks:
  - `word_count`
  - `sum_numbers` (integer inputs)
  - `hash_text`
  - `json_stats`

## Safety boundary

Do **not** put secrets, personal records, medical records, private messages, credentials, or other sensitive data into volunteer-computing jobs. Workers must be treated as untrusted machines. OIG v0.1.0 is for public, synthetic, sanitized, or otherwise safe-to-distribute inputs.

Read [SECURITY.md](SECURITY.md) and [docs/THREAT_MODEL.md](docs/THREAT_MODEL.md) before exposing a deployment to the internet.

## Quick start with Python

Requirements: Python 3.11 or newer.

```bash
python -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]'
cp .env.example .env
oig-server
```

Open:

- API documentation: `http://localhost:8000/docs`
- Minimal dashboard: `http://localhost:8000/`
- Health check: `http://localhost:8000/health`

The development admin key in `.env.example` is `change-me-now`. Change it before doing anything beyond local testing.

## Quick start with Docker

```bash
cp .env.example .env
docker compose up --build
```

The Compose setup starts PostgreSQL and the API.

## Run the demo

With the server running:

```bash
python examples/bootstrap_demo.py
```

The script creates an ontology, two workers, and a replicated word-count job. It prints two worker tokens. In two terminals, run:

```bash
oig-worker --server http://localhost:8000 --token WORKER_ONE_TOKEN --once
oig-worker --server http://localhost:8000 --token WORKER_TWO_TOKEN --once
```

Then inspect the job:

```bash
curl http://localhost:8000/v1/jobs/JOB_ID \
  -H 'X-Admin-Key: change-me-now'
```

## Repository map

```text
src/oig/
  api.py          HTTP routes
  config.py       Environment-based settings
  db.py           SQLAlchemy engine/session setup
  models.py       Persistent ontology, jobs, workers, and audit records
  schemas.py      Request and response models
  security.py     Admin and worker authentication
  services.py     Leasing, consensus, and graph services
  tasks.py        Safe deterministic worker task implementations
  worker.py       Volunteer worker client
  main.py         Application entry point and dashboard

tests/            API and consensus tests
docs/             Architecture, API, threat model, and roadmap
examples/         Runnable bootstrap example
```

## Guiding principles

- Public-interest uses before surveillance uses.
- Data minimization before data accumulation.
- Traceable transformations before black boxes.
- Compute moves toward sensitive data; sensitive data does not move toward strangers.
- No cryptocurrency is required for participation.
- Human governance is part of the architecture, not an afterthought.

## Current limitations

- The scheduler uses a relational database, not a peer-to-peer mesh.
- API-key authentication is suitable only for development and controlled pilots.
- There is no row-level policy engine yet.
- There are no signed job bundles or WebAssembly sandboxes yet.
- The ontology schema is stored and documented but not fully JSON-Schema validated.
- Database migrations are not yet included.
- Work leasing is deliberately simple and needs stronger transaction isolation at scale.

These limits are documented rather than hidden behind a cloud of enterprise incense.

## License

MIT. See [LICENSE](LICENSE).

## Working title

“Open Intelligence Grid” is a descriptive working title. Perform a proper trademark and project-name review before a major public launch.

================================================================================
FILE: SECURITY.md
================================================================================
# Security policy

## Status

OIG v0.1.0 is an alpha research and development release. It has not received a professional security audit and must not be used as-is for classified, regulated, medical, financial, or personally identifying information.

## Non-negotiable data rule

Volunteer workers are untrusted. Never distribute secrets or sensitive source records to them. Encryption in transit does not change this rule because a worker must ordinarily access a task's plaintext input to calculate its output.

## Supported deployment posture

For the alpha release:

- Keep the API behind a trusted reverse proxy.
- Use TLS.
- Replace the development admin key.
- Restrict administrative endpoints by network and identity.
- Use PostgreSQL for shared deployments.
- Log and monitor worker registrations and job submissions.
- Use only public, synthetic, or sanitized job inputs.
- Run trusted data processing on organization-controlled workers.

## Reporting vulnerabilities

Do not open a public issue containing exploit details or exposed data. Before public launch, create a dedicated security contact and private disclosure channel in the repository's `SECURITY.md`.

## Planned controls

See [docs/ROADMAP.md](docs/ROADMAP.md) for signed job bundles, scoped identities, policy enforcement, sandboxed execution, provenance signatures, and federation.

================================================================================
FILE: compose.yaml
================================================================================
services:
  db:
    image: postgres:17-alpine
    environment:
      POSTGRES_DB: oig
      POSTGRES_USER: oig
      POSTGRES_PASSWORD: oig-development-password
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U oig -d oig"]
      interval: 5s
      timeout: 5s
      retries: 10
    volumes:
      - oig-postgres:/var/lib/postgresql/data

  api:
    build: .
    depends_on:
      db:
        condition: service_healthy
    environment:
      OIG_ENVIRONMENT: development
      OIG_DATABASE_URL: postgresql+psycopg://oig:oig-development-password@db:5432/oig
      OIG_ADMIN_KEY: ${OIG_ADMIN_KEY:-change-me-now}
      OIG_LEASE_SECONDS: ${OIG_LEASE_SECONDS:-300}
      OIG_ALLOW_SAME_WORKER_REPLICAS: ${OIG_ALLOW_SAME_WORKER_REPLICAS:-false}
    ports:
      - "8000:8000"

volumes:
  oig-postgres:

================================================================================
FILE: docs/API.md
================================================================================
# API overview

Interactive OpenAPI documentation is available at `/docs` while the server is running.

## Authentication

Administrative endpoints require:

```http
X-Admin-Key: your-admin-key
```

Worker endpoints require:

```http
Authorization: Bearer worker-token
```

## Important endpoints

| Method | Path | Purpose |
|---|---|---|
| GET | `/health` | Liveness check |
| GET | `/v1/stats` | Counts and job status summary |
| POST | `/v1/object-types` | Create an ontology object type |
| POST | `/v1/entities` | Create an entity |
| POST | `/v1/relationships` | Create a relationship |
| GET | `/v1/entities/{id}/graph` | Read a one-hop graph |
| POST | `/v1/workers/register` | Register a worker and receive its token |
| POST | `/v1/jobs` | Create a replicated job |
| POST | `/v1/work/claim` | Lease work to an authenticated worker |
| POST | `/v1/work/{id}/submit` | Submit a completed result |
| POST | `/v1/work/{id}/fail` | Report a failed work unit |

## Create a job

```bash
curl -X POST http://localhost:8000/v1/jobs \
  -H 'Content-Type: application/json' \
  -H 'X-Admin-Key: change-me-now' \
  -d '{
    "task_type": "sum_numbers",
    "input_payload": {"numbers": [1, 2, 3, 4]},
    "replicas_required": 2,
    "max_replicas": 3
  }'
```

================================================================================
FILE: docs/ARCHITECTURE.md
================================================================================
# Architecture

## Current shape

```text
                         +-----------------------+
                         |   Browser / API user  |
                         +-----------+-----------+
                                     |
                                     v
+----------------+        +----------+-----------+        +----------------+
| Volunteer      | <----> | OIG Coordinator API | <----> | PostgreSQL or  |
| Worker A       |  HTTPS | FastAPI + services  |  SQL   | SQLite         |
+----------------+        +----------+-----------+        +----------------+
                                     ^
+----------------+                   |
| Volunteer      | <-----------------+
| Worker B       |
+----------------+
```

The coordinator is centralized in v0.1.0. Workers poll for jobs, lease a work unit, execute a known deterministic task, and submit a result. The coordinator accepts a job when enough independent workers produce the same canonical result hash.

## Ontology model

- `ObjectType` describes a class of real-world object and its expected properties.
- `Entity` is an instance of an object type.
- `Relationship` is a directed, typed edge between two entities.

The alpha stores property descriptions and entity properties as JSON. A later release should validate values against JSON Schema and add versioned ontology migrations.

## Job model

- `Job` contains a task type, input payload, validation threshold, and result.
- `WorkUnit` is one independently executed replica of a job.
- `Worker` has a bearer token, capabilities, activity state, and simple trust score.

A worker ordinarily cannot receive two replicas of the same job. Development deployments may override this with `OIG_ALLOW_SAME_WORKER_REPLICAS=true`.

## Consensus

Results are serialized as canonical JSON and hashed with SHA-256. A job completes when one hash has at least `replicas_required` completed replicas. If all issued replicas finish without consensus and the job has not reached `max_replicas`, the coordinator adds a tie-breaker work unit. Otherwise, the job becomes `disputed`.

This detects accidental and some malicious errors. It does not prove that a majority of workers are honest.

## Evolution path

1. Relational coordinator with trusted HTTPS workers.
2. Signed task manifests and sandboxed WebAssembly execution.
3. Organization-operated trusted worker pools.
4. Federated query and model execution near private data.
5. Replicated coordinators and append-only provenance logs.
6. Optional peer discovery and content-addressed task distribution.

Mesh networking belongs near the end because decentralizing an unclear protocol merely distributes the confusion.

================================================================================
FILE: docs/GITHUB_PUBLISHING.md
================================================================================
# Publishing this repository on GitHub

## 1. Review the working title

“Open Intelligence Grid” is a working title. Search existing projects and trademarks before promoting it as a permanent brand.

## 2. Create an empty repository

Create a new repository without automatically adding a README, license, or `.gitignore`, because those files already exist here.

## 3. Initialize and push

```bash
git init
git add .
git commit -m "Initial open-source alpha"
git branch -M main
git remote add origin YOUR_REPOSITORY_URL
git push -u origin main
```

## 4. Configure repository protections

Recommended settings:

- Require pull requests before merging to `main`.
- Require the CI workflow to pass.
- Require at least one approving review.
- Block force pushes and branch deletion.
- Enable secret scanning and dependency alerts where available.
- Create a private security-reporting channel before inviting broad use.

## 5. Create the first release

Tag the tested alpha:

```bash
git tag -a v0.1.0 -m "Open Intelligence Grid alpha"
git push origin v0.1.0
```

Attach the source archive and describe the current safety boundaries prominently. Do not present the alpha as suitable for sensitive or regulated data.

## 6. Suggested repository description

> Open-source ontology, data coordination, and independently validated volunteer-computing foundation.

## 7. Suggested initial topics

`open-source`, `distributed-computing`, `volunteer-computing`, `ontology`, `fastapi`, `postgresql`, `civic-tech`, `data-governance`

================================================================================
FILE: docs/QUICKSTART.md
================================================================================
# Quickstart

## 1. Install

```bash
python -m venv .venv
source .venv/bin/activate
pip install -e '.[dev]'
cp .env.example .env
```

## 2. Start the coordinator

```bash
oig-server
```

## 3. Bootstrap a demonstration

```bash
python examples/bootstrap_demo.py
```

Copy the two printed worker tokens and job ID.

## 4. Run two independent workers

Terminal A:

```bash
oig-worker --server http://localhost:8000 --token TOKEN_A --once
```

Terminal B:

```bash
oig-worker --server http://localhost:8000 --token TOKEN_B --once
```

## 5. Read the accepted job

```bash
curl http://localhost:8000/v1/jobs/JOB_ID \
  -H 'X-Admin-Key: change-me-now'
```

The job should have status `completed` and two matching completed work units.

================================================================================
FILE: docs/RELEASE_VALIDATION.md
================================================================================
# Release validation

Release: `0.1.0`

Validation performed on July 1, 2026 with Python 3.13.

## Automated checks

```text
pytest: 4 passed
ruff: all checks passed
compileall: passed
compose.yaml parse: passed
editable package installation: passed
```

## End-to-end check

A local coordinator was started with SQLite. The demonstration script then:

1. Created an ontology object type and entity.
2. Registered two distinct workers.
3. Created a word-count job requiring two matching replicas.
4. Ran each worker once through the command-line client.
5. Received two identical canonical result hashes.
6. Marked the job `completed` with the verified result.

## Important scope statement

These checks demonstrate that the alpha functions as designed in a local development environment. They are not a security audit, load test, formal protocol verification, or production-readiness certification.

================================================================================
FILE: docs/ROADMAP.md
================================================================================
# Roadmap

## Phase 1: Harden the alpha

- Alembic database migrations.
- Pagination and filtering.
- Idempotency keys.
- Rate limiting.
- Structured logs, metrics, and traces.
- Stronger transaction-safe work claiming.
- JSON Schema validation for ontology entities.
- Capability and resource matching.
- Worker heartbeat and health screens.

## Phase 2: Safe distributed execution

- Signed task manifests.
- Content-addressed input bundles.
- WebAssembly/WASI task sandbox.
- CPU, memory, disk, and time quotas.
- Network-disabled tasks by default.
- Reproducible task builds.
- Random known-answer challenge jobs.
- Weighted consensus and worker reputation.

## Phase 3: Institutional federation

- Organizations, projects, and scoped roles.
- OpenID Connect authentication.
- Open Policy Agent integration.
- Data residency policies.
- Trusted worker pools.
- Federated aggregate queries.
- Differential privacy options.
- Signed lineage and provenance export.

## Phase 4: Operational applications

- Ontology actions and approval workflows.
- Event streams and alerting.
- Geospatial object support.
- Notebook and SQL workspaces.
- Visual pipeline builder.
- Dashboard/application SDK.
- Connectors for public datasets and common databases.

## Phase 5: Resilient network

- Replicated coordinators.
- Append-only transparent audit log.
- Peer discovery.
- Content-addressed peer-to-peer distribution for public task bundles.
- Offline-capable edge nodes.
- Byzantine-fault research and formal protocol specification.

A full Palantir-class platform is a long-range ecosystem. Each phase must remain independently useful rather than serving as scaffolding for a castle that never opens its doors.

================================================================================
FILE: docs/THREAT_MODEL.md
================================================================================
# Threat model

## Assets

- Ontology definitions and entity data.
- Administrative credentials.
- Worker identities and tokens.
- Job inputs and accepted results.
- Audit history and provenance.
- Coordinator availability.

## Adversaries

- A malicious volunteer worker returning fabricated results.
- Several colluding workers.
- An attacker stealing an administrator key.
- A coordinator operator abusing access.
- A user submitting a harmful or privacy-invasive job.
- A denial-of-service attacker creating excessive jobs or leases.

## Existing alpha mitigations

- Worker tokens are stored as SHA-256 hashes, not plaintext.
- Work leases expire.
- Separate worker identities are required for replicated results by default.
- Accepted results require matching canonical hashes.
- Administrative writes require an admin key.
- Audit events record major writes and worker actions.

## Known gaps

- No fine-grained roles or organization boundaries.
- No rate limiting.
- No cryptographic signature on task definitions.
- No secure execution sandbox.
- No remote attestation.
- No Sybil resistance beyond controlled worker registration.
- No immutable or externally witnessed audit log.
- No privacy-budget accounting or differential privacy.
- No content safety review for submitted workloads.

## Recommended pilot rules

1. Permit only an allowlist of task implementations.
2. Permit only public or synthetic inputs.
3. Manually approve workers.
4. Place the coordinator behind TLS and an identity-aware proxy.
5. Monitor task volume and unusual result disagreements.
6. Keep a human appeal and correction process for operational uses.

================================================================================
FILE: examples/bootstrap_demo.py
================================================================================
import os

import httpx

SERVER = os.getenv("OIG_SERVER", "http://localhost:8000").rstrip("/")
ADMIN_KEY = os.getenv("OIG_ADMIN_KEY", "change-me-now")
HEADERS = {"X-Admin-Key": ADMIN_KEY}


def post(path: str, payload: dict) -> dict:
    response = httpx.post(f"{SERVER}{path}", headers=HEADERS, json=payload, timeout=30)
    response.raise_for_status()
    return response.json()


def main() -> None:
    organization = post(
        "/v1/object-types",
        {
            "name": "Organization",
            "description": "A community organization or institution.",
            "schema_json": {
                "type": "object",
                "properties": {"name": {"type": "string"}, "city": {"type": "string"}},
                "required": ["name"],
            },
        },
    )
    entity = post(
        "/v1/entities",
        {
            "object_type_id": organization["id"],
            "external_id": "demo-library",
            "properties": {"name": "Demo Public Library", "city": "Gloucester"},
        },
    )

    workers = []
    for name in ("demo-worker-a", "demo-worker-b"):
        workers.append(
            post(
                "/v1/workers/register",
                {"name": name, "capabilities": ["word_count", "sum_numbers", "hash_text", "json_stats"]},
            )
        )

    job = post(
        "/v1/jobs",
        {
            "task_type": "word_count",
            "input_payload": {
                "text": "Public knowledge should glow brighter when more hands carry the lantern."
            },
            "replicas_required": 2,
            "max_replicas": 3,
        },
    )

    print("Created entity:", entity["id"])
    print("Created job:", job["id"])
    for worker in workers:
        print(f"{worker['name']} token: {worker['token']}")


if __name__ == "__main__":
    main()

================================================================================
FILE: pyproject.toml
================================================================================
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "open-intelligence-grid"
version = "0.1.0"
description = "An open-source ontology and volunteer-computing coordination platform."
readme = "README.md"
requires-python = ">=3.11"
license = "MIT"
authors = [{ name = "Open Intelligence Grid contributors" }]
dependencies = [
  "fastapi>=0.115,<1.0",
  "uvicorn[standard]>=0.30,<1.0",
  "sqlalchemy>=2.0,<2.2",
  "pydantic>=2.8,<3.0",
  "pydantic-settings>=2.4,<3.0",
  "httpx>=0.27,<1.0",
  "psycopg[binary]>=3.2,<4.0",
]

[project.optional-dependencies]
dev = [
  "pytest>=8.0,<9.0",
  "pytest-cov>=5.0,<7.0",
  "ruff>=0.8,<1.0",
]

[project.scripts]
oig-server = "oig.main:run"
oig-worker = "oig.worker:main"

[tool.setuptools]
package-dir = {"" = "src"}

[tool.setuptools.packages.find]
where = ["src"]

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-q"

[tool.ruff]
line-length = 100
target-version = "py311"

================================================================================
FILE: src/oig/__init__.py
================================================================================
"""Open Intelligence Grid core package."""

__version__ = "0.1.0"

================================================================================
FILE: src/oig/api.py
================================================================================
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, selectinload

from .db import get_db
from .models import AuditEvent, Entity, Job, ObjectType, Relationship, Worker, WorkUnit
from .schemas import (
    AuditRead,
    EntityCreate,
    EntityRead,
    GraphRead,
    JobCreate,
    JobRead,
    ObjectTypeCreate,
    ObjectTypeRead,
    RelationshipCreate,
    RelationshipRead,
    WorkerRead,
    WorkerRegister,
    WorkerRegistered,
    WorkClaim,
    WorkFailure,
    WorkSubmit,
)
from .security import hash_token, issue_token, require_admin, require_worker
from .services import audit, create_job, fail_work, graph_for_entity, lease_work, submit_result
from .tasks import TASKS

router = APIRouter(prefix="/v1")
Admin = Annotated[str, Depends(require_admin)]
DB = Annotated[Session, Depends(get_db)]
AuthenticatedWorker = Annotated[Worker, Depends(require_worker)]


@router.get("/stats")
def stats(db: DB) -> dict:
    job_rows = db.execute(select(Job.status, func.count(Job.id)).group_by(Job.status)).all()
    return {
        "object_types": db.scalar(select(func.count(ObjectType.id))) or 0,
        "entities": db.scalar(select(func.count(Entity.id))) or 0,
        "relationships": db.scalar(select(func.count(Relationship.id))) or 0,
        "workers": db.scalar(select(func.count(Worker.id))) or 0,
        "jobs": dict(job_rows),
        "supported_tasks": sorted(TASKS),
    }


@router.post("/object-types", response_model=ObjectTypeRead, status_code=status.HTTP_201_CREATED)
def create_object_type(payload: ObjectTypeCreate, db: DB, _admin: Admin) -> ObjectType:
    item = ObjectType(**payload.model_dump(by_alias=True))
    db.add(item)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="object_type.created",
        target_type="object_type",
        target_id=item.id,
        details={"name": item.name},
    )
    try:
        db.commit()
    except IntegrityError as exc:
        db.rollback()
        raise HTTPException(status_code=409, detail="Object type name already exists") from exc
    db.refresh(item)
    return item


@router.get("/object-types", response_model=list[ObjectTypeRead])
def list_object_types(db: DB, limit: int = Query(default=100, ge=1, le=500)) -> list[ObjectType]:
    return list(db.scalars(select(ObjectType).order_by(ObjectType.name).limit(limit)).all())


@router.post("/entities", response_model=EntityRead, status_code=status.HTTP_201_CREATED)
def create_entity(payload: EntityCreate, db: DB, _admin: Admin) -> Entity:
    if db.get(ObjectType, payload.object_type_id) is None:
        raise HTTPException(status_code=404, detail="Object type not found")
    entity = Entity(**payload.model_dump())
    db.add(entity)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="entity.created",
        target_type="entity",
        target_id=entity.id,
        details={"object_type_id": entity.object_type_id},
    )
    try:
        db.commit()
    except IntegrityError as exc:
        db.rollback()
        raise HTTPException(status_code=409, detail="Duplicate external ID for object type") from exc
    db.refresh(entity)
    return entity


@router.get("/entities", response_model=list[EntityRead])
def list_entities(
    db: DB,
    object_type_id: str | None = None,
    limit: int = Query(default=100, ge=1, le=500),
) -> list[Entity]:
    statement = select(Entity).order_by(Entity.created_at.desc()).limit(limit)
    if object_type_id:
        statement = statement.where(Entity.object_type_id == object_type_id)
    return list(db.scalars(statement).all())


@router.get("/entities/{entity_id}", response_model=EntityRead)
def get_entity(entity_id: str, db: DB) -> Entity:
    entity = db.get(Entity, entity_id)
    if entity is None:
        raise HTTPException(status_code=404, detail="Entity not found")
    return entity


@router.post(
    "/relationships", response_model=RelationshipRead, status_code=status.HTTP_201_CREATED
)
def create_relationship(payload: RelationshipCreate, db: DB, _admin: Admin) -> Relationship:
    if db.get(Entity, payload.source_entity_id) is None:
        raise HTTPException(status_code=404, detail="Source entity not found")
    if db.get(Entity, payload.target_entity_id) is None:
        raise HTTPException(status_code=404, detail="Target entity not found")
    relationship = Relationship(**payload.model_dump())
    db.add(relationship)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="relationship.created",
        target_type="relationship",
        target_id=relationship.id,
        details={"relationship_type": relationship.relationship_type},
    )
    db.commit()
    db.refresh(relationship)
    return relationship


@router.get("/relationships", response_model=list[RelationshipRead])
def list_relationships(db: DB, limit: int = Query(default=100, ge=1, le=500)) -> list[Relationship]:
    return list(
        db.scalars(select(Relationship).order_by(Relationship.created_at.desc()).limit(limit)).all()
    )


@router.get("/entities/{entity_id}/graph", response_model=GraphRead)
def entity_graph(entity_id: str, db: DB) -> GraphRead:
    graph = graph_for_entity(db, entity_id)
    if graph is None:
        raise HTTPException(status_code=404, detail="Entity not found")
    return graph


@router.post("/workers/register", response_model=WorkerRegistered, status_code=status.HTTP_201_CREATED)
def register_worker(payload: WorkerRegister, db: DB, _admin: Admin) -> WorkerRegistered:
    token = issue_token()
    worker = Worker(
        name=payload.name,
        capabilities=payload.capabilities,
        token_hash=hash_token(token),
    )
    db.add(worker)
    db.flush()
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="worker.registered",
        target_type="worker",
        target_id=worker.id,
        details={"name": worker.name, "capabilities": worker.capabilities},
    )
    db.commit()
    db.refresh(worker)
    return WorkerRegistered(
        id=worker.id,
        name=worker.name,
        capabilities=worker.capabilities,
        token=token,
    )


@router.get("/workers", response_model=list[WorkerRead])
def list_workers(db: DB, _admin: Admin) -> list[Worker]:
    return list(db.scalars(select(Worker).order_by(Worker.created_at.desc())).all())


@router.post("/jobs", response_model=JobRead, status_code=status.HTTP_201_CREATED)
def submit_job(payload: JobCreate, db: DB, _admin: Admin) -> Job:
    if payload.task_type not in TASKS:
        raise HTTPException(
            status_code=400,
            detail=f"Unsupported task type. Supported: {', '.join(sorted(TASKS))}",
        )
    return create_job(db, **payload.model_dump())


@router.get("/jobs", response_model=list[JobRead])
def list_jobs(
    db: DB,
    _admin: Admin,
    limit: int = Query(default=100, ge=1, le=500),
) -> list[Job]:
    return list(
        db.scalars(
            select(Job)
            .options(selectinload(Job.work_units))
            .order_by(Job.created_at.desc())
            .limit(limit)
        ).all()
    )


@router.get("/jobs/{job_id}", response_model=JobRead)
def get_job(job_id: str, db: DB, _admin: Admin) -> Job:
    job = db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job_id)
    )
    if job is None:
        raise HTTPException(status_code=404, detail="Job not found")
    return job


@router.post("/work/claim", response_model=WorkClaim | None)
def claim_work(db: DB, worker: AuthenticatedWorker, response: Response) -> WorkClaim | None:
    unit = lease_work(db, worker)
    if unit is None:
        response.status_code = status.HTTP_204_NO_CONTENT
        return None
    return WorkClaim(
        work_unit_id=unit.id,
        lease_token=unit.lease_token or "",
        lease_expires_at=unit.lease_expires_at,
        job_id=unit.job.id,
        task_type=unit.job.task_type,
        input_payload=unit.job.input_payload,
    )


@router.post("/work/{work_unit_id}/submit", response_model=JobRead)
def submit_work_result(
    work_unit_id: str,
    payload: WorkSubmit,
    db: DB,
    worker: AuthenticatedWorker,
) -> Job:
    unit = db.get(WorkUnit, work_unit_id)
    if unit is None:
        raise HTTPException(status_code=404, detail="Work unit not found")
    try:
        return submit_result(
            db,
            unit=unit,
            worker=worker,
            lease_token=payload.lease_token,
            result_payload=payload.result_payload,
        )
    except ValueError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@router.post("/work/{work_unit_id}/fail", response_model=JobRead)
def report_work_failure(
    work_unit_id: str,
    payload: WorkFailure,
    db: DB,
    worker: AuthenticatedWorker,
) -> Job:
    unit = db.get(WorkUnit, work_unit_id)
    if unit is None:
        raise HTTPException(status_code=404, detail="Work unit not found")
    try:
        return fail_work(
            db,
            unit=unit,
            worker=worker,
            lease_token=payload.lease_token,
            error=payload.error,
        )
    except ValueError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@router.get("/audit", response_model=list[AuditRead])
def list_audit_events(
    db: DB,
    _admin: Admin,
    limit: int = Query(default=100, ge=1, le=500),
) -> list[AuditEvent]:
    return list(
        db.scalars(select(AuditEvent).order_by(AuditEvent.created_at.desc()).limit(limit)).all()
    )

================================================================================
FILE: src/oig/config.py
================================================================================
from functools import lru_cache

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="OIG_",
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    app_name: str = "Open Intelligence Grid"
    environment: str = "development"
    database_url: str = "sqlite:///./oig.db"
    admin_key: str = "change-me-now"
    lease_seconds: int = 300
    allow_same_worker_replicas: bool = False
    cors_origins: list[str] = Field(default_factory=list)


@lru_cache
def get_settings() -> Settings:
    return Settings()

================================================================================
FILE: src/oig/db.py
================================================================================
from collections.abc import Generator

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

from .config import get_settings


class Base(DeclarativeBase):
    pass


_settings = get_settings()
_connect_args = {"check_same_thread": False} if _settings.database_url.startswith("sqlite") else {}
engine = create_engine(_settings.database_url, connect_args=_connect_args, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)


def init_db() -> None:
    from . import models  # noqa: F401

    Base.metadata.create_all(bind=engine)


def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

================================================================================
FILE: src/oig/main.py
================================================================================
from contextlib import asynccontextmanager

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse

from .api import router
from .config import get_settings
from .db import init_db


DASHBOARD = """
<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>Open Intelligence Grid</title>
  <style>
    :root { color-scheme: dark; font-family: ui-monospace, SFMono-Regular, Menlo, monospace; }
    body { margin: 0; background: #101419; color: #e8f0e8; }
    main { max-width: 960px; margin: 0 auto; padding: 48px 24px; }
    h1 { font-size: clamp(2rem, 6vw, 4.5rem); margin: 0; letter-spacing: -0.06em; }
    .subtitle { color: #9db0a2; max-width: 720px; line-height: 1.6; }
    .grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(150px, 1fr)); gap: 12px; margin: 32px 0; }
    .card { border: 1px solid #334139; border-radius: 12px; padding: 18px; background: #151c19; }
    .number { font-size: 2rem; font-weight: 700; }
    a { color: #a9ffbe; }
    pre { white-space: pre-wrap; word-break: break-word; background: #0b0e0c; padding: 16px; border-radius: 12px; }
    .pulse { display: inline-block; width: 10px; height: 10px; border-radius: 50%; background: #a9ffbe; box-shadow: 0 0 16px #a9ffbe; }
  </style>
</head>
<body>
<main>
  <div><span class="pulse"></span> alpha coordinator online</div>
  <h1>Open Intelligence Grid</h1>
  <p class="subtitle">An inspectable ontology and distributed-computing foundation. Small enough to understand. Large enough to begin.</p>
  <div class="grid" id="stats"><div class="card">Loading statistics...</div></div>
  <p><a href="/docs">Interactive API documentation</a> · <a href="/health">Health check</a></p>
  <h2>Supported worker tasks</h2>
  <pre id="tasks">Loading...</pre>
</main>
<script>
fetch('/v1/stats').then(r => r.json()).then(data => {
  const entries = [
    ['Object types', data.object_types], ['Entities', data.entities],
    ['Relationships', data.relationships], ['Workers', data.workers],
    ['Jobs', Object.values(data.jobs).reduce((a,b) => a+b, 0)]
  ];
  document.getElementById('stats').innerHTML = entries.map(([k,v]) => `<div class="card"><div class="number">${v}</div><div>${k}</div></div>`).join('');
  document.getElementById('tasks').textContent = data.supported_tasks.join('\n');
}).catch(error => {
  document.getElementById('stats').innerHTML = `<div class="card">Could not load stats: ${error}</div>`;
});
</script>
</body>
</html>
"""


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_db()
    yield


settings = get_settings()
app = FastAPI(
    title=settings.app_name,
    version="0.1.0",
    description="Open ontology and distributed volunteer-computing coordinator.",
    lifespan=lifespan,
)
if settings.cors_origins:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=False,
        allow_methods=["GET", "POST", "PATCH", "DELETE"],
        allow_headers=["Authorization", "Content-Type", "X-Admin-Key"],
    )
app.include_router(router)


@app.get("/", response_class=HTMLResponse, include_in_schema=False)
def dashboard() -> str:
    return DASHBOARD


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok", "version": "0.1.0", "environment": settings.environment}


def run() -> None:
    uvicorn.run("oig.main:app", host="0.0.0.0", port=8000, reload=False)


if __name__ == "__main__":
    run()

================================================================================
FILE: src/oig/models.py
================================================================================
from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, JSON, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship

from .db import Base


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


def uuid_string() -> str:
    return str(uuid.uuid4())


class ObjectType(Base):
    __tablename__ = "object_types"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    name: Mapped[str] = mapped_column(String(120), unique=True, index=True)
    description: Mapped[str] = mapped_column(Text, default="")
    schema_json: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)

    entities: Mapped[list[Entity]] = relationship(back_populates="object_type")


class Entity(Base):
    __tablename__ = "entities"
    __table_args__ = (UniqueConstraint("object_type_id", "external_id", name="uq_entity_external"),)

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    object_type_id: Mapped[str] = mapped_column(ForeignKey("object_types.id"), index=True)
    external_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
    properties: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow)

    object_type: Mapped[ObjectType] = relationship(back_populates="entities")


class Relationship(Base):
    __tablename__ = "relationships"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    relationship_type: Mapped[str] = mapped_column(String(120), index=True)
    source_entity_id: Mapped[str] = mapped_column(ForeignKey("entities.id"), index=True)
    target_entity_id: Mapped[str] = mapped_column(ForeignKey("entities.id"), index=True)
    properties: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)

    source: Mapped[Entity] = relationship(foreign_keys=[source_entity_id])
    target: Mapped[Entity] = relationship(foreign_keys=[target_entity_id])


class Worker(Base):
    __tablename__ = "workers"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    name: Mapped[str] = mapped_column(String(160))
    token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True)
    capabilities: Mapped[list[str]] = mapped_column(JSON, default=list)
    trust_score: Mapped[float] = mapped_column(Float, default=1.0)
    active: Mapped[bool] = mapped_column(Boolean, default=True)
    last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)


class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    task_type: Mapped[str] = mapped_column(String(120), index=True)
    input_payload: Mapped[dict[str, Any]] = mapped_column(JSON)
    status: Mapped[str] = mapped_column(String(30), default="queued", index=True)
    replicas_required: Mapped[int] = mapped_column(Integer, default=2)
    max_replicas: Mapped[int] = mapped_column(Integer, default=3)
    result_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
    result_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow)

    work_units: Mapped[list[WorkUnit]] = relationship(
        back_populates="job", cascade="all, delete-orphan", order_by="WorkUnit.created_at"
    )


class WorkUnit(Base):
    __tablename__ = "work_units"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid_string)
    job_id: Mapped[str] = mapped_column(ForeignKey("jobs.id"), index=True)
    replica_index: Mapped[int] = mapped_column(Integer)
    status: Mapped[str] = mapped_column(String(30), default="queued", index=True)
    worker_id: Mapped[str | None] = mapped_column(ForeignKey("workers.id"), nullable=True, index=True)
    lease_token: Mapped[str | None] = mapped_column(String(64), nullable=True)
    lease_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    result_payload: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
    result_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
    error: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    job: Mapped[Job] = relationship(back_populates="work_units")
    worker: Mapped[Worker | None] = relationship()


class AuditEvent(Base):
    __tablename__ = "audit_events"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    actor_type: Mapped[str] = mapped_column(String(40))
    actor_id: Mapped[str] = mapped_column(String(160))
    action: Mapped[str] = mapped_column(String(160), index=True)
    target_type: Mapped[str] = mapped_column(String(80))
    target_id: Mapped[str] = mapped_column(String(160))
    details: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, index=True)

================================================================================
FILE: src/oig/schemas.py
================================================================================
from datetime import datetime
from typing import Any

from pydantic import BaseModel, ConfigDict, Field, model_validator


class ORMModel(BaseModel):
    model_config = ConfigDict(from_attributes=True)


class ObjectTypeCreate(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    name: str = Field(min_length=1, max_length=120, pattern=r"^[A-Za-z][A-Za-z0-9_-]*$")
    description: str = ""
    schema_definition: dict[str, Any] = Field(default_factory=dict, alias="schema_json")


class ObjectTypeRead(ORMModel):
    id: str
    name: str
    description: str
    schema_definition: dict[str, Any] = Field(
        validation_alias="schema_json", serialization_alias="schema_json"
    )
    created_at: datetime


class EntityCreate(BaseModel):
    object_type_id: str
    external_id: str | None = None
    properties: dict[str, Any] = Field(default_factory=dict)


class EntityRead(ORMModel):
    id: str
    object_type_id: str
    external_id: str | None
    properties: dict[str, Any]
    created_at: datetime
    updated_at: datetime


class RelationshipCreate(BaseModel):
    relationship_type: str = Field(min_length=1, max_length=120)
    source_entity_id: str
    target_entity_id: str
    properties: dict[str, Any] = Field(default_factory=dict)


class RelationshipRead(ORMModel):
    id: str
    relationship_type: str
    source_entity_id: str
    target_entity_id: str
    properties: dict[str, Any]
    created_at: datetime


class GraphRead(BaseModel):
    center: EntityRead
    entities: list[EntityRead]
    relationships: list[RelationshipRead]


class WorkerRegister(BaseModel):
    name: str = Field(min_length=1, max_length=160)
    capabilities: list[str] = Field(default_factory=list)


class WorkerRegistered(BaseModel):
    id: str
    name: str
    capabilities: list[str]
    token: str


class WorkerRead(ORMModel):
    id: str
    name: str
    capabilities: list[str]
    trust_score: float
    active: bool
    last_seen_at: datetime | None
    created_at: datetime


class JobCreate(BaseModel):
    task_type: str = Field(min_length=1, max_length=120)
    input_payload: dict[str, Any]
    replicas_required: int = Field(default=2, ge=1, le=10)
    max_replicas: int = Field(default=3, ge=1, le=20)

    @model_validator(mode="after")
    def validate_replica_counts(self) -> "JobCreate":
        if self.max_replicas < self.replicas_required:
            raise ValueError("max_replicas must be greater than or equal to replicas_required")
        return self


class WorkUnitRead(ORMModel):
    id: str
    job_id: str
    replica_index: int
    status: str
    worker_id: str | None
    lease_expires_at: datetime | None
    result_hash: str | None
    error: str | None
    created_at: datetime
    completed_at: datetime | None


class JobRead(ORMModel):
    id: str
    task_type: str
    input_payload: dict[str, Any]
    status: str
    replicas_required: int
    max_replicas: int
    result_payload: dict[str, Any] | None
    result_hash: str | None
    created_at: datetime
    updated_at: datetime
    work_units: list[WorkUnitRead] = Field(default_factory=list)


class WorkClaim(BaseModel):
    work_unit_id: str
    lease_token: str
    lease_expires_at: datetime
    job_id: str
    task_type: str
    input_payload: dict[str, Any]


class WorkSubmit(BaseModel):
    lease_token: str
    result_payload: dict[str, Any]


class WorkFailure(BaseModel):
    lease_token: str
    error: str = Field(min_length=1, max_length=4000)


class AuditRead(ORMModel):
    id: int
    actor_type: str
    actor_id: str
    action: str
    target_type: str
    target_id: str
    details: dict[str, Any]
    created_at: datetime

================================================================================
FILE: src/oig/security.py
================================================================================
import hashlib
import secrets

from fastapi import Depends, Header, HTTPException, status
from sqlalchemy import select
from sqlalchemy.orm import Session

from .config import get_settings
from .db import get_db
from .models import Worker, utcnow


def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def issue_token() -> str:
    return secrets.token_urlsafe(32)


def require_admin(x_admin_key: str = Header(default="")) -> str:
    settings = get_settings()
    if not x_admin_key or not secrets.compare_digest(x_admin_key, settings.admin_key):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid admin key")
    return "admin"


def require_worker(
    authorization: str = Header(default=""), db: Session = Depends(get_db)
) -> Worker:
    scheme, _, token = authorization.partition(" ")
    if scheme.lower() != "bearer" or not token:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Bearer token required")

    worker = db.scalar(select(Worker).where(Worker.token_hash == hash_token(token)))
    if worker is None or not worker.active:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid worker token")

    worker.last_seen_at = utcnow()
    db.commit()
    db.refresh(worker)
    return worker

================================================================================
FILE: src/oig/services.py
================================================================================
from __future__ import annotations

import hashlib
import json
import secrets
from collections import Counter
from datetime import datetime, timedelta, timezone
from typing import Any

from sqlalchemy import or_, select
from sqlalchemy.orm import Session, selectinload

from .config import get_settings
from .models import AuditEvent, Entity, Job, Relationship, Worker, WorkUnit, utcnow
from .schemas import GraphRead


def _as_utc(value: datetime) -> datetime:
    """Normalize timestamps returned by SQLite, which may drop timezone metadata."""
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)


def canonical_hash(payload: dict[str, Any]) -> str:
    encoded = json.dumps(
        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def audit(
    db: Session,
    *,
    actor_type: str,
    actor_id: str,
    action: str,
    target_type: str,
    target_id: str,
    details: dict[str, Any] | None = None,
) -> None:
    db.add(
        AuditEvent(
            actor_type=actor_type,
            actor_id=actor_id,
            action=action,
            target_type=target_type,
            target_id=target_id,
            details=details or {},
        )
    )


def create_job(db: Session, *, task_type: str, input_payload: dict[str, Any], replicas_required: int, max_replicas: int) -> Job:
    job = Job(
        task_type=task_type,
        input_payload=input_payload,
        replicas_required=replicas_required,
        max_replicas=max_replicas,
    )
    db.add(job)
    db.flush()
    for replica_index in range(replicas_required):
        db.add(WorkUnit(job_id=job.id, replica_index=replica_index))
    audit(
        db,
        actor_type="admin",
        actor_id="admin",
        action="job.created",
        target_type="job",
        target_id=job.id,
        details={"task_type": task_type, "replicas_required": replicas_required},
    )
    db.commit()
    return db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job.id)
    )


def _release_expired_leases(db: Session) -> None:
    now = utcnow()
    expired = db.scalars(
        select(WorkUnit).where(
            WorkUnit.status == "leased",
            WorkUnit.lease_expires_at.is_not(None),
            WorkUnit.lease_expires_at < now,
        )
    ).all()
    for unit in expired:
        unit.status = "queued"
        unit.worker_id = None
        unit.lease_token = None
        unit.lease_expires_at = None
    if expired:
        db.commit()


def lease_work(db: Session, worker: Worker) -> WorkUnit | None:
    settings = get_settings()
    _release_expired_leases(db)

    candidates = db.scalars(
        select(WorkUnit)
        .options(selectinload(WorkUnit.job))
        .where(WorkUnit.status == "queued")
        .order_by(WorkUnit.created_at)
    ).all()

    for unit in candidates:
        job = unit.job
        if job.status not in {"queued", "running"}:
            continue
        if worker.capabilities and job.task_type not in worker.capabilities:
            continue
        if not settings.allow_same_worker_replicas:
            already_worked = db.scalar(
                select(WorkUnit.id).where(
                    WorkUnit.job_id == job.id,
                    WorkUnit.worker_id == worker.id,
                )
            )
            if already_worked is not None:
                continue

        unit.status = "leased"
        unit.worker_id = worker.id
        unit.lease_token = secrets.token_urlsafe(24)
        unit.lease_expires_at = utcnow() + timedelta(seconds=settings.lease_seconds)
        job.status = "running"
        audit(
            db,
            actor_type="worker",
            actor_id=worker.id,
            action="work.leased",
            target_type="work_unit",
            target_id=unit.id,
            details={"job_id": job.id},
        )
        db.commit()
        db.refresh(unit)
        return unit
    return None


def _evaluate_job(db: Session, job: Job) -> None:
    completed = [unit for unit in job.work_units if unit.status == "completed" and unit.result_hash]
    counts = Counter(unit.result_hash for unit in completed)

    if counts:
        winning_hash, winning_count = counts.most_common(1)[0]
        if winning_count >= job.replicas_required:
            winning_unit = next(unit for unit in completed if unit.result_hash == winning_hash)
            job.status = "completed"
            job.result_hash = winning_hash
            job.result_payload = winning_unit.result_payload
            for unit in job.work_units:
                if unit.status == "queued":
                    unit.status = "cancelled"
            audit(
                db,
                actor_type="system",
                actor_id="consensus",
                action="job.completed",
                target_type="job",
                target_id=job.id,
                details={"result_hash": winning_hash, "matching_replicas": winning_count},
            )
            return

    active_or_waiting = [unit for unit in job.work_units if unit.status in {"queued", "leased"}]
    if active_or_waiting:
        return

    if len(job.work_units) < job.max_replicas:
        next_index = max((unit.replica_index for unit in job.work_units), default=-1) + 1
        job.work_units.append(WorkUnit(job_id=job.id, replica_index=next_index))
        job.status = "running"
        audit(
            db,
            actor_type="system",
            actor_id="consensus",
            action="work.tie_breaker_created",
            target_type="job",
            target_id=job.id,
            details={"replica_index": next_index},
        )
    else:
        job.status = "disputed"
        audit(
            db,
            actor_type="system",
            actor_id="consensus",
            action="job.disputed",
            target_type="job",
            target_id=job.id,
            details={"hash_counts": dict(counts)},
        )


def submit_result(
    db: Session,
    *,
    unit: WorkUnit,
    worker: Worker,
    lease_token: str,
    result_payload: dict[str, Any],
) -> Job:
    if unit.status != "leased" or unit.worker_id != worker.id:
        raise ValueError("Work unit is not leased to this worker")
    if not unit.lease_token or not secrets.compare_digest(unit.lease_token, lease_token):
        raise ValueError("Invalid lease token")
    if unit.lease_expires_at and _as_utc(unit.lease_expires_at) < utcnow():
        raise ValueError("Lease expired")

    unit.status = "completed"
    unit.result_payload = result_payload
    unit.result_hash = canonical_hash(result_payload)
    unit.completed_at = utcnow()
    unit.lease_token = None
    unit.lease_expires_at = None
    worker.trust_score = min(worker.trust_score + 0.01, 10.0)
    audit(
        db,
        actor_type="worker",
        actor_id=worker.id,
        action="work.completed",
        target_type="work_unit",
        target_id=unit.id,
        details={"result_hash": unit.result_hash},
    )

    db.flush()
    job = db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == unit.job_id)
    )
    if job is None:
        raise ValueError("Parent job not found")
    _evaluate_job(db, job)
    db.commit()
    return db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job.id)
    )


def fail_work(
    db: Session, *, unit: WorkUnit, worker: Worker, lease_token: str, error: str
) -> Job:
    if unit.status != "leased" or unit.worker_id != worker.id:
        raise ValueError("Work unit is not leased to this worker")
    if not unit.lease_token or not secrets.compare_digest(unit.lease_token, lease_token):
        raise ValueError("Invalid lease token")

    unit.status = "failed"
    unit.error = error
    unit.completed_at = utcnow()
    unit.lease_token = None
    unit.lease_expires_at = None
    worker.trust_score = max(worker.trust_score - 0.02, 0.0)

    db.flush()
    job = db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == unit.job_id)
    )
    if job is None:
        raise ValueError("Parent job not found")
    _evaluate_job(db, job)
    db.commit()
    return db.scalar(
        select(Job).options(selectinload(Job.work_units)).where(Job.id == job.id)
    )


def graph_for_entity(db: Session, entity_id: str) -> GraphRead | None:
    center = db.get(Entity, entity_id)
    if center is None:
        return None
    relationships = db.scalars(
        select(Relationship).where(
            or_(
                Relationship.source_entity_id == entity_id,
                Relationship.target_entity_id == entity_id,
            )
        )
    ).all()
    neighbor_ids = {
        rel.target_entity_id if rel.source_entity_id == entity_id else rel.source_entity_id
        for rel in relationships
    }
    neighbors = db.scalars(select(Entity).where(Entity.id.in_(neighbor_ids))).all() if neighbor_ids else []
    return GraphRead(center=center, entities=[center, *neighbors], relationships=relationships)

================================================================================
FILE: src/oig/tasks.py
================================================================================
import hashlib
import json
from collections.abc import Callable
from typing import Any


class TaskError(ValueError):
    pass


def word_count(payload: dict[str, Any]) -> dict[str, Any]:
    text = payload.get("text")
    if not isinstance(text, str):
        raise TaskError("word_count requires a string field named 'text'")
    return {
        "words": len(text.split()),
        "characters": len(text),
        "lines": len(text.splitlines()) if text else 0,
    }


def sum_numbers(payload: dict[str, Any]) -> dict[str, Any]:
    numbers = payload.get("numbers")
    if not isinstance(numbers, list) or not all(
        isinstance(value, int) and not isinstance(value, bool) for value in numbers
    ):
        raise TaskError("sum_numbers requires an integer list field named 'numbers'")
    return {"sum": sum(numbers), "count": len(numbers)}


def hash_text(payload: dict[str, Any]) -> dict[str, Any]:
    text = payload.get("text")
    algorithm = payload.get("algorithm", "sha256")
    if not isinstance(text, str):
        raise TaskError("hash_text requires a string field named 'text'")
    if algorithm not in {"sha256", "sha512", "blake2b"}:
        raise TaskError("algorithm must be sha256, sha512, or blake2b")
    digest = hashlib.new(algorithm, text.encode("utf-8")).hexdigest()
    return {"algorithm": algorithm, "digest": digest}


def json_stats(payload: dict[str, Any]) -> dict[str, Any]:
    def walk(value: Any) -> tuple[int, int, int, int]:
        if isinstance(value, dict):
            child = [walk(item) for item in value.values()]
            return (
                1 + sum(item[0] for item in child),
                sum(item[1] for item in child),
                sum(item[2] for item in child),
                sum(item[3] for item in child),
            )
        if isinstance(value, list):
            child = [walk(item) for item in value]
            return (
                sum(item[0] for item in child),
                1 + sum(item[1] for item in child),
                sum(item[2] for item in child),
                sum(item[3] for item in child),
            )
        if isinstance(value, (str, int, float, bool)) or value is None:
            return (0, 0, 1, len(json.dumps(value, ensure_ascii=False)))
        raise TaskError("payload contains an unsupported JSON value")

    objects, arrays, scalars, scalar_bytes = walk(payload)
    return {
        "objects": objects,
        "arrays": arrays,
        "scalars": scalars,
        "serialized_scalar_characters": scalar_bytes,
    }


TASKS: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = {
    "word_count": word_count,
    "sum_numbers": sum_numbers,
    "hash_text": hash_text,
    "json_stats": json_stats,
}


def run_task(task_type: str, payload: dict[str, Any]) -> dict[str, Any]:
    task = TASKS.get(task_type)
    if task is None:
        raise TaskError(f"Unsupported task type: {task_type}")
    return task(payload)

================================================================================
FILE: src/oig/worker.py
================================================================================
import argparse
import sys
import time

import httpx

from .tasks import TaskError, run_task


def worker_headers(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def process_once(server: str, token: str, timeout: float = 30.0) -> bool:
    server = server.rstrip("/")
    with httpx.Client(timeout=timeout, headers=worker_headers(token)) as client:
        claim = client.post(f"{server}/v1/work/claim")
        if claim.status_code == 204:
            return False
        claim.raise_for_status()
        work = claim.json()

        try:
            result = run_task(work["task_type"], work["input_payload"])
        except (TaskError, Exception) as exc:
            failure = client.post(
                f"{server}/v1/work/{work['work_unit_id']}/fail",
                json={"lease_token": work["lease_token"], "error": str(exc)},
            )
            failure.raise_for_status()
            print(f"Failed work unit {work['work_unit_id']}: {exc}", file=sys.stderr)
            return True

        response = client.post(
            f"{server}/v1/work/{work['work_unit_id']}/submit",
            json={"lease_token": work["lease_token"], "result_payload": result},
        )
        response.raise_for_status()
        job = response.json()
        print(
            f"Completed work unit {work['work_unit_id']} for job {work['job_id']} "
            f"(job status: {job['status']})"
        )
        return True


def main() -> None:
    parser = argparse.ArgumentParser(description="Open Intelligence Grid volunteer worker")
    parser.add_argument("--server", default="http://localhost:8000")
    parser.add_argument("--token", required=True, help="Worker token returned during registration")
    parser.add_argument("--once", action="store_true", help="Claim at most one work unit and exit")
    parser.add_argument("--poll-seconds", type=float, default=10.0)
    args = parser.parse_args()

    while True:
        try:
            worked = process_once(args.server, args.token)
        except httpx.HTTPError as exc:
            print(f"Coordinator request failed: {exc}", file=sys.stderr)
            worked = False

        if args.once:
            return
        if not worked:
            time.sleep(max(args.poll_seconds, 1.0))


if __name__ == "__main__":
    main()

================================================================================
FILE: tests/conftest.py
================================================================================
from pathlib import Path

import pytest
from fastapi.testclient import TestClient


@pytest.fixture()
def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    database_path = tmp_path / "test.db"
    monkeypatch.setenv("OIG_DATABASE_URL", f"sqlite:///{database_path}")
    monkeypatch.setenv("OIG_ADMIN_KEY", "test-admin")
    monkeypatch.setenv("OIG_ALLOW_SAME_WORKER_REPLICAS", "false")

    import oig.config as config

    config.get_settings.cache_clear()

    import oig.db as db

    db.engine.dispose()
    connect_args = {"check_same_thread": False}
    db.engine = db.create_engine(
        f"sqlite:///{database_path}", connect_args=connect_args, pool_pre_ping=True
    )
    db.SessionLocal.configure(bind=db.engine)

    from oig.main import app

    with TestClient(app) as test_client:
        yield test_client

================================================================================
FILE: tests/test_health.py
================================================================================
def test_health(client):
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "ok"

================================================================================
FILE: tests/test_jobs.py
================================================================================
ADMIN = {"X-Admin-Key": "test-admin"}


def register_worker(client, name):
    response = client.post(
        "/v1/workers/register",
        headers=ADMIN,
        json={"name": name, "capabilities": ["sum_numbers"]},
    )
    assert response.status_code == 201
    return response.json()


def claim(client, token):
    response = client.post(
        "/v1/work/claim", headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 200
    return response.json()


def submit(client, token, work, result):
    return client.post(
        f"/v1/work/{work['work_unit_id']}/submit",
        headers={"Authorization": f"Bearer {token}"},
        json={"lease_token": work["lease_token"], "result_payload": result},
    )


def test_two_workers_reach_consensus(client):
    worker_a = register_worker(client, "a")
    worker_b = register_worker(client, "b")

    job = client.post(
        "/v1/jobs",
        headers=ADMIN,
        json={
            "task_type": "sum_numbers",
            "input_payload": {"numbers": [1, 2, 3]},
            "replicas_required": 2,
            "max_replicas": 3,
        },
    )
    assert job.status_code == 201

    work_a = claim(client, worker_a["token"])
    work_b = claim(client, worker_b["token"])

    result_a = submit(client, worker_a["token"], work_a, {"sum": 6, "count": 3})
    result_b = submit(client, worker_b["token"], work_b, {"count": 3, "sum": 6})
    assert result_a.status_code == 200
    assert result_b.status_code == 200
    assert result_b.json()["status"] == "completed"
    assert result_b.json()["result_payload"] == {"count": 3, "sum": 6}


def test_disagreement_creates_tie_breaker(client):
    workers = [register_worker(client, name) for name in ("a", "b", "c")]
    client.post(
        "/v1/jobs",
        headers=ADMIN,
        json={
            "task_type": "sum_numbers",
            "input_payload": {"numbers": [4, 5]},
            "replicas_required": 2,
            "max_replicas": 3,
        },
    ).json()

    first = claim(client, workers[0]["token"])
    second = claim(client, workers[1]["token"])
    submit(client, workers[0]["token"], first, {"sum": 9, "count": 2})
    disputed_so_far = submit(client, workers[1]["token"], second, {"sum": 99, "count": 2})
    assert disputed_so_far.status_code == 200
    assert len(disputed_so_far.json()["work_units"]) == 3

    tie_breaker = claim(client, workers[2]["token"])
    final = submit(client, workers[2]["token"], tie_breaker, {"count": 2, "sum": 9})
    assert final.status_code == 200
    assert final.json()["status"] == "completed"
    assert final.json()["result_payload"]["sum"] == 9

================================================================================
FILE: tests/test_ontology.py
================================================================================
ADMIN = {"X-Admin-Key": "test-admin"}


def test_create_entities_and_graph(client):
    org_type = client.post(
        "/v1/object-types",
        headers=ADMIN,
        json={"name": "Organization", "schema_json": {"type": "object"}},
    )
    assert org_type.status_code == 201

    program_type = client.post(
        "/v1/object-types",
        headers=ADMIN,
        json={"name": "Program", "schema_json": {"type": "object"}},
    )
    assert program_type.status_code == 201

    org = client.post(
        "/v1/entities",
        headers=ADMIN,
        json={"object_type_id": org_type.json()["id"], "properties": {"name": "Library"}},
    )
    program = client.post(
        "/v1/entities",
        headers=ADMIN,
        json={"object_type_id": program_type.json()["id"], "properties": {"name": "Robotics"}},
    )
    assert org.status_code == 201
    assert program.status_code == 201

    relation = client.post(
        "/v1/relationships",
        headers=ADMIN,
        json={
            "relationship_type": "OPERATES",
            "source_entity_id": org.json()["id"],
            "target_entity_id": program.json()["id"],
            "properties": {},
        },
    )
    assert relation.status_code == 201

    graph = client.get(f"/v1/entities/{org.json()['id']}/graph")
    assert graph.status_code == 200
    assert len(graph.json()["entities"]) == 2
    assert graph.json()["relationships"][0]["relationship_type"] == "OPERATES"

 

Back to blog

Leave a comment

Please note, comments need to be approved before they are published.