Scenario 1 — Hello-World through the Knowledge Graph#
The smallest complete picture of operation coordination: one middleware instance offers a capability, another asks for that capability to be carried out, and the knowledge graph is the only thing connecting them. Neither instance is ever given the other’s address.
Two resources take part. A hello resource registers a hello_world workflow and then waits. A planner wants that work done. The planner holds no URL for the hello resource — it creates an Operation in the graph, and finds the hello resource’s event trigger by asking the graph which Service realizes the capability it needs. The hello resource pulls the queued Operation, runs it, and writes the outcome back. What remains afterwards is a permanent, queryable record of what was asked, which workflow did it, and what came back.
Scenario 2 shows the other half of the picture: direct invocation, with no queue at all.
Before you run this. A GraphDB must be reachable, with
GRAPHDB_URL,GRAPHDB_USERNAMEandGRAPHDB_PASSWORDset to point at it. Every cell below works against thekapps-demorepository, which is named in the code rather than read from the environment — aGRAPHDB_REPOSITORYleft in your environment is ignored. Step 1 erases that repository completely, so make sureGRAPHDB_URLnames a server you are allowed to overwrite.
import logging
import threading
import time
import uvicorn
from rdflib.namespace import RDF
from handlers import hello_world
from kapps_ogm import OGM
from kapps_semantic_middleware import SemanticMiddleware
from kapps_semantic_middleware.credentials import DEMO_REPOSITORY, graphdb_for
from kapps_semantic_middleware.registration import mint_capability_iri, mint_workflow_iri
from kapps_semantic_middleware.vocabulary import CFC, OperationStatus, SVC
import seed
# The libraries above log at INFO to stderr. A stderr stream renders as an error block on the
# documentation site, and it is grouped ahead of this notebook's own stdout -- so the HTTP
# request lines would appear above the step that made them. The narrative here is the print
# output; keep third-party logging at WARNING so that is what you read.
for _name in ('', 'httpx', 'httpcore', 'GraphDB', 'KafkaManager', 'kapps_ogm'):
logging.getLogger(_name).setLevel(logging.WARNING)
def serve(middleware, port):
"""Start `middleware` on a background thread and wait until it has registered.
The thread is what lets this notebook keep executing cells while a server runs, and it is
also what makes `stop` below mandatory. Uvicorn installs signal handlers only on the main
thread, so off it SIGTERM never reaches the ASGI lifespan and the middleware's on_shutdown
callbacks -- deregistration among them -- never run. `stop` sets should_exit and joins,
which runs that same lifespan shutdown with no signal involved. Copy this helper and you
inherit both halves: a copy that never calls `stop` leaks a Service on every run.
"""
config = uvicorn.Config(middleware.app, host='127.0.0.1', port=port, log_level='warning')
server = uvicorn.Server(config)
thread = threading.Thread(target=server.run, daemon=True)
thread.start()
t0 = time.time()
while not server.started and time.time() - t0 < 30:
time.sleep(0.05)
if not server.started:
raise RuntimeError(f'server on port {port} did not start in time')
return server, thread
def stop(server, thread):
"""Stop a served middleware so its deregistration callbacks run."""
server.should_exit = True
thread.join(timeout=20)
db = graphdb_for(DEMO_REPOSITORY)
print(f"Connected to GraphDB repository: {db.repository}")
Connected to GraphDB repository: kapps-demo
Step 1 — Seed a Clean Repository#
seed_scenario1 does three things in order: it clears the repository, loads the ontologies this scenario needs, and creates the two resource individuals the rest of the notebook refers to.
The order matters. The middleware never invents ontology classes at runtime — the Service, Capability and Workflow classes it registers against must already exist in the graph before an instance starts. Seeding is how they get there in an example; in a real deployment the ontology is loaded once and the resources are provisioned separately.
Clearing first is what makes the notebook re-runnable. Every run starts from the same empty repository, so a half-finished earlier run cannot leave a stale svc:address behind to confuse discovery.
seed.seed_scenario1(db)
print('Hello resource: ', db.triple_exists((seed.HELLO_RESOURCE, RDF.type, seed.HELLO_RESOURCE_CLASS)))
print('Planner resource:', db.triple_exists((seed.PLANNER_RESOURCE, RDF.type, seed.PLANNER_RESOURCE_CLASS)))
Hello resource: True
Planner resource: True
Step 2 — Start the Hello-World Middleware#
Two things happen here, and only the second one touches the graph.
SemanticMiddleware(mode='resource', ...) binds one instance to one Resource. The workflow(...) decorator then declares that hello_world realizes a Capability of that Resource. At this point nothing has been written anywhere: the declaration is held in memory.
Starting the server is what registers. On startup the instance writes its Service individual, the Capability, the Workflow, and the reachability triples that tell other peers where to find it. This is why an instance that is constructed but never served does not exist as far as the graph is concerned — and why the planner in step 4 is served too.
The serve helper defined at the top of this notebook runs uvicorn on a background thread so that the notebook can keep executing cells. That choice has a cost, and the helper’s docstring explains it: on a background thread uvicorn installs no signal handlers, so the middleware’s shutdown callbacks never fire on their own. Step 6 stops the servers explicitly, which is what makes deregistration run. Skip step 6 and you leave an address in the graph for a kernel that is gone.
mw1 = SemanticMiddleware(
mode='resource', resource_iri=seed.HELLO_RESOURCE,
service_class=seed.HELLO_SERVICE_CLASS, ogm=OGM(db=db),
host='127.0.0.1', port=8993,
)
mw1.workflow(capability_class=seed.HELLO_CAPABILITY_CLASS,
workflow_class=seed.HELLO_WORKFLOW_CLASS)(hello_world)
server, thread = serve(mw1, 8993)
print('Hello middleware started on port 8993')
print('Routes:', [r.path for r in mw1.app.routes if 'workflows' in r.path])
Hello middleware started on port 8993
Routes: ['/workflows/event_trigger/execute', '/workflows/event_trigger/execute_background', '/workflows/event_trigger/description', '/workflows/event_trigger/interrupt', '/workflows/hello_world/execute', '/workflows/hello_world/execute_background', '/workflows/hello_world/description', '/workflows/hello_world/interrupt']
Step 3 — Inspect What Registration Wrote#
Registration wrote a small, deliberately shaped subgraph. Three structural triples say how the pieces belong together: the Service belongs to the Resource (isServiceOf), the Workflow belongs to the Service (isWorkflowOf), and the Capability is realized by the Workflow (realizedByWorkflow).
Only the instance-owned direction of each pair is stored. The graph does not also carry hasService, hasWorkflow and so on — a reasoner can derive them, and materializing both directions would mean two places to keep correct. Query in the direction that is written.
Two further triples are the ones that matter for discovery. svc:address is the Service’s base URL, and svc:endpoint is the directly callable URL of one workflow. Both are written on startup and removed on shutdown, which makes them a liveness signal rather than configuration: an address in the graph means something answered there recently.
Note that the Service IRI is read off the instance rather than rebuilt from the Resource IRI. One Resource may carry several Services — one per running instance — so the IRI carries a discriminator and there is nothing to reconstruct.
# Per-instance — read off the instance rather than rebuilt from the resource.
service_iri = mw1.service_iri
cap_iri = mint_capability_iri(seed.HELLO_RESOURCE, 'hello_world')
wf_iri = mint_workflow_iri(service_iri, 'hello_world')
assert db.triple_exists((service_iri, SVC.isServiceOf, seed.HELLO_RESOURCE))
assert db.triple_exists((cap_iri, SVC.realizedByWorkflow, wf_iri))
assert db.triple_exists((wf_iri, SVC.isWorkflowOf, service_iri))
print('address: ', list(db.triples_get(sub=service_iri, pred=SVC.address)))
print('endpoint:', list(db.triples_get(sub=wf_iri, pred=SVC.endpoint)))
address: [(IRI('https://example.org/kapps-demo#hello_resource_service_http_c__s__s_127_d_0_d_0_d_1_c_8993'), IRI('https://w3id.org/circularfactory/Service#address'), 'http://127.0.0.1:8993')]
endpoint: [(IRI('https://example.org/kapps-demo#hello_resource_service_http_c__s__s_127_d_0_d_0_d_1_c_8993_workflow_hello_world'), IRI('https://w3id.org/circularfactory/Service#endpoint'), 'http://127.0.0.1:8993/workflows/hello_world/execute')]
Step 4 — Dispatch through the Event Trigger, then Pull-and-Run#
This is the whole coordination pattern in one cell.
The planner is a middleware instance in its own right, and it is served, not merely constructed. A resource-mode instance that never runs never registers, so it would hold an ogm and nothing else. A peer that hands out work is discoverable like any other, and the hello resource could ring it back.
planner.request(...) opens a transaction. On exit it creates the Operation in the graph with status queued, addressed at the Capability rather than at a machine. It then rings the hello resource’s event trigger over REST — and it finds that trigger by querying the graph, which is the only place the planner learns where the hello resource is.
The event trigger is a notification, not the work itself. It tells the hello resource that something is waiting. claim_next() is the other half: the hello resource takes the queued Operation, runs the handler, and the context manager records the result. Work is pulled by the resource that will do it, never pushed into it, so a resource that is busy or offline simply has not claimed yet and the Operation stays queued.
planner = SemanticMiddleware(
mode='resource', resource_iri=seed.PLANNER_RESOURCE,
service_class=seed.PLANNER_SERVICE_CLASS, ogm=OGM(db=graphdb_for(DEMO_REPOSITORY)),
host='127.0.0.1', port=8994,
)
# The planner is served, not merely constructed. on_start_up is what writes its Service
# individual and its svc:address, so an unserved resource-mode instance makes mode='resource' a
# label with no runtime consequence -- and this notebook is meant to be copied.
planner_server, planner_thread = serve(planner, 8994)
planner_service = planner.service_iri
print('Planner served on port 8994')
print(' planner address in graph:', list(db.triples_get(sub=planner_service, pred=SVC.address)))
with planner.request(capability_class=seed.HELLO_CAPABILITY_CLASS,
operation_class=str(CFC.Operation)) as op:
pass # hello_world takes no arguments
op_iri = op.iri
print('Planner dispatched operation:', op_iri)
with mw1.claim_next() as claimed:
claimed.result = hello_world()
print('Hello resource pulled and ran it -> result:', repr(claimed.result))
Planner served on port 8994
planner address in graph: [(IRI('https://example.org/kapps-demo#planner_resource_service_http_c__s__s_127_d_0_d_0_d_1_c_8994'), IRI('https://w3id.org/circularfactory/Service#address'), 'http://127.0.0.1:8994')]
Planner dispatched operation: https://w3id.org/circularfactory/Core#Operation_op_0c89d067abc5
Hello resource pulled and ran it -> result: 'hello world'
Step 5 — The Decision is Now Traceable in the Graph#
The Operation that step 4 created is still there, and it now carries its own outcome: the terminal status, the workflow that executed it, when that happened, and what it returned.
The status is the provenance record. There is no separate success flag and no log line to correlate — the terminal transition wrote the status and the execution facts in one commit, so an Operation cannot be found done without the evidence of what did it. Anything that can query the graph can audit this afterwards, including peers that were not running at the time.
print('operationStatus: ', list(db.triples_get(sub=op_iri, pred=SVC.operationStatus)))
print('executedByWorkflow:', list(db.triples_get(sub=op_iri, pred=SVC.executedByWorkflow)))
print('executionResult: ', list(db.triples_get(sub=op_iri, pred=SVC.executionResult)))
assert list(db.triples_get(sub=op_iri, pred=SVC.operationStatus))[0][2] == OperationStatus.DONE
operationStatus: [(IRI('https://w3id.org/circularfactory/Core#Operation_op_0c89d067abc5'), IRI('https://w3id.org/circularfactory/Service#operationStatus'), 'done')]
executedByWorkflow: [(IRI('https://w3id.org/circularfactory/Core#Operation_op_0c89d067abc5'), IRI('https://w3id.org/circularfactory/Service#executedByWorkflow'), IRI('https://example.org/kapps-demo#hello_resource_service_http_c__s__s_127_d_0_d_0_d_1_c_8993_workflow_hello_world'))]
executionResult: [(IRI('https://w3id.org/circularfactory/Core#Operation_op_0c89d067abc5'), IRI('https://w3id.org/circularfactory/Service#executionResult'), 'hello world')]
Step 6 — Shutdown and Deregistration#
Stopping the servers runs the shutdown callbacks, and those callbacks remove exactly the triples that advertise reachability: each Service’s address and the Workflow’s endpoint. Both servers are stopped, because both registered.
Everything else stays. The Service, Capability and Workflow individuals remain in the graph, as does the finished Operation. This is the distinction the model rests on: what a resource is is durable, and where it can be reached right now is not. A peer querying after this cell finds the hello resource described in full and finds no address on it, which is exactly the truth — it exists, and it is not running.
If you interrupt the kernel instead of running this cell, that cleanup does not happen and the next run starts against a graph with a stale address in it. Step 1 clears the repository, which is why re-running the notebook from the top recovers by itself.
stop(server, thread)
stop(planner_server, planner_thread)
time.sleep(0.5)
print('hello address removed: ', not list(db.triples_get(sub=service_iri, pred=SVC.address)))
print('workflow endpoint removed:', not list(db.triples_get(sub=wf_iri, pred=SVC.endpoint)))
print('planner address removed: ', not list(db.triples_get(sub=planner_service, pred=SVC.address)))
print('workflow individual preserved:', db.triple_exists((wf_iri, RDF.type, seed.HELLO_WORKFLOW_CLASS)))
print('planner Service preserved: ', db.triple_exists((planner_service, SVC.isServiceOf, seed.PLANNER_RESOURCE)))
hello address removed: True
workflow endpoint removed: True
planner address removed: True
workflow individual preserved: True
planner Service preserved: True