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:
- Register a worker
- Receive work
- Process the work independently
- Submit a result
- Compare that result against another worker’s result
- Reach verified consensus
- Detect disagreement
- 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"