Merge remote-tracking branch 'origin/main' into codex/wp-400-graph-runtime

Reviewed-on: #109

adapters/agentbridge/http.go

@@ -0,0 +1,61 @@
+package agentbridge
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+
+	"git.cin.su/m/Assistant/contracts/work"
+)
+
+var ErrUnavailable = errors.New("agent execution bridge unavailable")
+
+type HTTPExecutor struct {
+	URL    string
+	Client *http.Client
+}
+
+func NewHTTPExecutor(url string, client *http.Client) (*HTTPExecutor, error) {
+	url = strings.TrimSpace(url)
+	if url == "" || !strings.HasPrefix(url, "http://") && !strings.HasPrefix(url, "https://") {
+		return nil, ErrUnavailable
+	}
+	if client == nil {
+		client = http.DefaultClient
+	}
+	return &HTTPExecutor{URL: strings.TrimRight(url, "/"), Client: client}, nil
+}
+
+func (e *HTTPExecutor) Execute(ctx context.Context, request work.AgentExecutionRequest) (work.AgentExecutionResult, error) {
+	body, err := json.Marshal(request)
+	if err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	httpRequest, err := http.NewRequestWithContext(ctx, http.MethodPost, e.URL+"/v1/execute", bytes.NewReader(body))
+	if err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	httpRequest.Header.Set("Content-Type", "application/json")
+	response, err := e.Client.Do(httpRequest)
+	if err != nil {
+		return work.AgentExecutionResult{}, fmt.Errorf("%w: %v", ErrUnavailable, err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode < 200 || response.StatusCode >= 300 {
+		return work.AgentExecutionResult{}, fmt.Errorf("%w: status=%s", ErrUnavailable, response.Status)
+	}
+	decoder := json.NewDecoder(io.LimitReader(response.Body, 1<<20))
+	var result work.AgentExecutionResult
+	if err := decoder.Decode(&result); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	if result.State == "" || result.Summary == "" || result.ObservedAtUnixMillis < 1 {
+		return work.AgentExecutionResult{}, errors.New("agent execution bridge returned invalid result")
+	}
+	return result, nil
+}

adapters/agentbridge/http_test.go

@@ -0,0 +1,43 @@
+package agentbridge
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"git.cin.su/m/Assistant/contracts/work"
+)
+
+func TestHTTPExecutorCarriesTypedDispatchAndDecodesResult(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost || r.URL.Path != "/v1/execute" {
+			t.Fatalf("unexpected request: %s %s", r.Method, r.URL.Path)
+		}
+		var request work.AgentExecutionRequest
+		if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
+			t.Fatal(err)
+		}
+		if request.WorkerID != "worker-1" || request.FencingToken != 7 || request.Intent.DispatchID != "dispatch-1" {
+			t.Fatalf("request=%+v", request)
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(work.AgentExecutionResult{State: work.AgentExecutionCompleted, Summary: "completed", ObservedAtUnixMillis: 2})
+	}))
+	defer server.Close()
+	executor, err := NewHTTPExecutor(server.URL, server.Client())
+	if err != nil {
+		t.Fatal(err)
+	}
+	result, err := executor.Execute(context.Background(), work.AgentExecutionRequest{Intent: work.AgentDispatchIntent{DispatchID: "dispatch-1"}, WorkerID: "worker-1", FencingToken: 7})
+	if err != nil || result.State != work.AgentExecutionCompleted || result.Summary != "completed" {
+		t.Fatalf("result=%+v err=%v", result, err)
+	}
+}
+
+func TestHTTPExecutorRejectsUnavailableEndpoint(t *testing.T) {
+	if _, err := NewHTTPExecutor("", nil); err != ErrUnavailable {
+		t.Fatalf("expected unavailable, got %v", err)
+	}
+}

adapters/postgres/epistemics_repository.go

@@ -21,6 +21,10 @@ func OpenEpistemics(ctx context.Context, config ContinuityConfig) (*EpistemicsRe
 
 func (r *EpistemicsRepository) Close() { r.pool.Close() }
 
+func (r *EpistemicsRepository) Ping(ctx context.Context) error {
+	return r.pool.Ping(ctx)
+}
+
 func (r *EpistemicsRepository) SaveSystemExpectation(ctx context.Context, value epistemics.SystemExpectation) error {
 	_, err := r.pool.Exec(ctx, `INSERT INTO epistemics.system_expectations (
 		expectation_id, original_intent_id, source_utterance_id, expected_outcome, expected_effect,
@@ -161,3 +165,21 @@ func (r *EpistemicsRepository) SaveReflectionCandidate(ctx context.Context, valu
 		value.Summary, value.EvidenceRefs, value.RequiresCognition, value.RecordedAtUnixMillis, value.Revision)
 	return err
 }
+
+func (r *EpistemicsRepository) SaveTruthBundle(ctx context.Context, value epistemics.TruthBundle, recordedAtUnixMillis int64) error {
+	if value.Source.ID == "" || value.Snapshot.ID == "" || value.Excerpt.ID == "" || value.Claim.ID == "" || value.Belief.ID == "" || value.Citation.ID == "" || recordedAtUnixMillis < 1 {
+		return fmt.Errorf("invalid truth bundle")
+	}
+	_, err := r.pool.Exec(ctx, `INSERT INTO epistemics.truth_bundles (
+		bundle_id,source_id,source_kind,source_locator,source_title,source_publisher,snapshot_id,snapshot_source_id,snapshot_content_digest,snapshot_captured_at_unix_millis,snapshot_freshness,
+		excerpt_id,excerpt_snapshot_id,excerpt_text,excerpt_start,excerpt_end,observation_id,observation_observed_at_unix_millis,claim_id,claim_statement,claim_observation_id,claim_citation_id,claim_verification,
+		belief_id,belief_claim_id,belief_state,citation_id,citation_excerpt_id,citation_evidence,citation_coverage,citation_conflict,citation_resolved_at_unix_millis,recorded_at_unix_millis)
+		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31,$32,$33)
+		ON CONFLICT (bundle_id) DO UPDATE SET recorded_at_unix_millis=EXCLUDED.recorded_at_unix_millis`,
+		"bundle:"+string(value.Citation.ID), value.Source.ID, value.Source.Kind, value.Source.Locator, value.Source.Title, value.Source.Publisher,
+		value.Snapshot.ID, value.Snapshot.SourceID, value.Snapshot.ContentDigest, value.Snapshot.CapturedAtUnixMillis, value.Snapshot.Freshness,
+		value.Excerpt.ID, value.Excerpt.SnapshotID, value.Excerpt.Text, value.Excerpt.Start, value.Excerpt.End,
+		value.Observation.ID, value.Observation.ObservedAt, value.Claim.ID, value.Claim.Statement, value.Claim.ObservationID, value.Claim.CitationID, value.Claim.Verification,
+		value.Belief.ID, value.Belief.ClaimID, value.Belief.State, value.Citation.ID, value.Citation.ExcerptID, value.Citation.Evidence, value.Citation.Coverage, value.Citation.Conflict, value.Citation.ResolvedAtUnixMillis, recordedAtUnixMillis)
+	return err
+}

adapters/postgres/work_repository.go

@@ -26,7 +26,8 @@ func OpenWork(ctx context.Context, config ContinuityConfig) (*WorkRepository, er
 	}
 	return &WorkRepository{pool}, nil
 }
-func (r *WorkRepository) Close() { r.pool.Close() }
+func (r *WorkRepository) Close()                         { r.pool.Close() }
+func (r *WorkRepository) Ping(ctx context.Context) error { return r.pool.Ping(ctx) }
 func (r *WorkRepository) Create(ctx context.Context, target work.TaskTarget) error {
 	_, err := r.pool.Exec(ctx, `INSERT INTO work.tasks (task_id,goal_id,scope_id,revision) VALUES ($1,$2,$3,$4) ON CONFLICT (task_id) DO NOTHING`, target.TaskID, target.GoalID, target.ScopeID, target.Revision)
 	return err
@@ -143,7 +144,7 @@ func (r *WorkRepository) SaveWorkGraph(ctx context.Context, graph work.WorkGraph
 		return err
 	}
 	for _, node := range graph.Nodes {
-		if _, err := tx.Exec(ctx, `INSERT INTO work.graph_nodes(graph_id,revision,node_id,idempotency_key,state,attempts) VALUES($1,$2,$3,$4,$5,$6) ON CONFLICT(graph_id,revision,node_id) DO NOTHING`, graph.ID, graph.Revision, node.ID, node.IdempotencyKey, node.State, node.Attempts); err != nil {
+		if _, err := tx.Exec(ctx, `INSERT INTO work.graph_nodes(graph_id,revision,node_id,dispatch_id,idempotency_key,state,attempts) VALUES($1,$2,$3,$4,$5,$6,$7) ON CONFLICT(graph_id,revision,node_id) DO NOTHING`, graph.ID, graph.Revision, node.ID, node.DispatchID, node.IdempotencyKey, node.State, node.Attempts); err != nil {
 			return err
 		}
 	}
@@ -161,13 +162,13 @@ func (r *WorkRepository) LoadWorkGraph(ctx context.Context, graphID string, revi
 	if err != nil {
 		return work.WorkGraph{}, fmt.Errorf("load graph: %w", err)
 	}
-	rows, err := r.pool.Query(ctx, `SELECT node_id,idempotency_key,state,attempts FROM work.graph_nodes WHERE graph_id=$1 AND revision=$2 ORDER BY node_id`, graphID, revision)
+	rows, err := r.pool.Query(ctx, `SELECT node_id,dispatch_id,idempotency_key,state,attempts FROM work.graph_nodes WHERE graph_id=$1 AND revision=$2 ORDER BY node_id`, graphID, revision)
 	if err != nil {
 		return work.WorkGraph{}, err
 	}
 	for rows.Next() {
 		var node work.GraphNode
-		if err := rows.Scan(&node.ID, &node.IdempotencyKey, &node.State, &node.Attempts); err != nil {
+		if err := rows.Scan(&node.ID, &node.DispatchID, &node.IdempotencyKey, &node.State, &node.Attempts); err != nil {
 			rows.Close()
 			return work.WorkGraph{}, err
 		}

adapters/research/http_connector.go

@@ -0,0 +1,74 @@
+package research
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+
+	"git.cin.su/m/Assistant/contracts/epistemics"
+)
+
+var ErrHTTPSourceRejected = errors.New("http research source rejected")
+
+type HTTPConnector struct {
+	Client       *http.Client
+	AllowedHosts map[string]struct{}
+	MaxBytes     int64
+}
+
+func (c HTTPConnector) Collect(ctx context.Context, request epistemics.WebResearchRequest, nowUnixMillis int64) (epistemics.ResearchCandidate, error) {
+	if request.Plan.Validate() != nil || request.SourceID == "" || strings.TrimSpace(request.Statement) == "" || nowUnixMillis < 1 {
+		return epistemics.ResearchCandidate{}, ErrHTTPSourceRejected
+	}
+	u, err := url.Parse(request.URL)
+	if err != nil || u.Scheme != "https" || u.User != nil || u.Host == "" || !c.allowedHost(strings.ToLower(u.Hostname())) {
+		return epistemics.ResearchCandidate{}, ErrHTTPSourceRejected
+	}
+	client := c.Client
+	if client == nil {
+		client = &http.Client{}
+	}
+	maxBytes := c.MaxBytes
+	if maxBytes <= 0 || maxBytes > 4<<20 {
+		maxBytes = 256 << 10
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
+	if err != nil {
+		return epistemics.ResearchCandidate{}, ErrHTTPSourceRejected
+	}
+	response, err := client.Do(req)
+	if err != nil {
+		return epistemics.ResearchCandidate{}, fmt.Errorf("read web source: %w", err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode < 200 || response.StatusCode >= 300 {
+		return epistemics.ResearchCandidate{}, fmt.Errorf("web source status: %s", response.Status)
+	}
+	body, err := io.ReadAll(io.LimitReader(response.Body, maxBytes+1))
+	if err != nil || int64(len(body)) > maxBytes || len(body) == 0 {
+		return epistemics.ResearchCandidate{}, ErrHTTPSourceRejected
+	}
+	digest := sha256.Sum256(body)
+	digestText := "sha256:" + hex.EncodeToString(digest[:])
+	key := sha256.Sum256([]byte(u.String() + "\x00" + digestText))
+	keyText := hex.EncodeToString(key[:])
+	snapshotID := epistemics.SourceSnapshotID("snapshot:" + keyText)
+	return epistemics.ResearchCandidate{
+		PropositionKey: request.PropositionKey, Statement: request.Statement,
+		Source:   epistemics.Source{ID: request.SourceID, Kind: "web", Locator: u.String(), Title: request.Title, Publisher: u.Hostname()},
+		Snapshot: epistemics.SourceSnapshot{ID: snapshotID, SourceID: request.SourceID, ContentDigest: digestText, CapturedAtUnixMillis: nowUnixMillis, Freshness: epistemics.FreshnessCurrent},
+		Excerpt:  epistemics.SourceExcerpt{ID: epistemics.SourceExcerptID("excerpt:" + keyText), SnapshotID: snapshotID, Text: string(body), Start: 0, End: int64(len(body))},
+		Evidence: epistemics.EvidenceObserved, Coverage: epistemics.CoverageComplete, Conflict: epistemics.ConflictNone,
+	}, nil
+}
+
+func (c HTTPConnector) allowedHost(host string) bool {
+	_, ok := c.AllowedHosts[host]
+	return ok
+}

adapters/research/repository_connector.go

@@ -0,0 +1,63 @@
+package research
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"git.cin.su/m/Assistant/contracts/epistemics"
+)
+
+var ErrRepositoryPathRejected = errors.New("repository source path rejected")
+
+type RepositoryConnector struct{ Root string }
+
+func (c RepositoryConnector) Collect(ctx context.Context, request epistemics.RepositoryResearchRequest, nowUnixMillis int64) (epistemics.ResearchCandidate, error) {
+	if err := request.Plan.Validate(); err != nil {
+		return epistemics.ResearchCandidate{}, err
+	}
+	if request.SourceID == "" || strings.TrimSpace(request.Statement) == "" || nowUnixMillis < 1 {
+		return epistemics.ResearchCandidate{}, ErrRepositoryPathRejected
+	}
+	root, err := filepath.Abs(c.Root)
+	if err != nil || root == "" {
+		return epistemics.ResearchCandidate{}, ErrRepositoryPathRejected
+	}
+	path, err := filepath.Abs(filepath.Join(root, filepath.FromSlash(request.Path)))
+	if err != nil || !withinRoot(root, path) {
+		return epistemics.ResearchCandidate{}, ErrRepositoryPathRejected
+	}
+	select {
+	case <-ctx.Done():
+		return epistemics.ResearchCandidate{}, ctx.Err()
+	default:
+	}
+	content, err := os.ReadFile(path)
+	if err != nil || len(content) == 0 {
+		return epistemics.ResearchCandidate{}, fmt.Errorf("read repository source: %w", err)
+	}
+	digest := sha256.Sum256(content)
+	digestText := "sha256:" + hex.EncodeToString(digest[:])
+	key := sha256.Sum256([]byte(filepath.ToSlash(path) + "\x00" + digestText))
+	keyText := hex.EncodeToString(key[:])
+	return epistemics.ResearchCandidate{
+		PropositionKey: request.PropositionKey,
+		Statement:      request.Statement,
+		Source:         epistemics.Source{ID: request.SourceID, Kind: "repository", Locator: "file://" + filepath.ToSlash(path), Title: filepath.Base(path), Publisher: "local-repository"},
+		Snapshot:       epistemics.SourceSnapshot{ID: epistemics.SourceSnapshotID("snapshot:" + keyText), SourceID: request.SourceID, ContentDigest: digestText, CapturedAtUnixMillis: nowUnixMillis, Freshness: epistemics.FreshnessCurrent},
+		Excerpt:        epistemics.SourceExcerpt{ID: epistemics.SourceExcerptID("excerpt:" + keyText), SnapshotID: epistemics.SourceSnapshotID("snapshot:" + keyText), Text: string(content), Start: 0, End: int64(len(content))},
+		Evidence:       epistemics.EvidenceObserved,
+		Coverage:       epistemics.CoverageComplete,
+		Conflict:       epistemics.ConflictNone,
+	}, nil
+}
+
+func withinRoot(root, path string) bool {
+	rel, err := filepath.Rel(root, path)
+	return err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) && !filepath.IsAbs(rel)
+}

cmd/agent-worker/main.go

@@ -5,13 +5,16 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
+	"net/http"
 	"os"
 	"time"
 
+	"git.cin.su/m/Assistant/adapters/agentbridge"
 	"git.cin.su/m/Assistant/adapters/postgres"
 	secretcontracts "git.cin.su/m/Assistant/contracts/secrets"
 	"git.cin.su/m/Assistant/contracts/work"
 	"git.cin.su/m/Assistant/modules/work/application"
+	"git.cin.su/m/Assistant/modules/work/ports"
 	platformsecrets "git.cin.su/m/Assistant/platform/secrets"
 )
 
@@ -34,9 +37,25 @@ func main() {
 		os.Exit(1)
 	}
 	defer repository.Close()
+	var executor ports.AgentExecutor
+	if bridgeURL := os.Getenv("AGENT_EXECUTION_BRIDGE_URL"); bridgeURL != "" {
+		executor, err = agentbridge.NewHTTPExecutor(bridgeURL, &http.Client{Timeout: *leaseTTL})
+		if err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			os.Exit(1)
+		}
+	} else {
+		fmt.Fprintln(os.Stderr, "agent execution bridge is not configured; using legacy dry-run dispatch")
+	}
 	ctx := context.Background()
 	for {
-		result, cycleErr := application.ExecuteWorkerCycle(ctx, repository, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL)
+		var result application.WorkerCycleResult
+		var cycleErr error
+		if executor == nil {
+			result, cycleErr = application.ExecuteWorkerCycle(ctx, repository, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL)
+		} else {
+			result, cycleErr = application.ExecuteWorkerCycleWithExecutor(ctx, repository, executor, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL)
+		}
 		if cycleErr != nil {
 			fmt.Fprintln(os.Stderr, cycleErr)
 			if *once {

cmd/cognitive-worker/fleet_status.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"git.cin.su/m/Assistant/contracts/capability"
+	fleetdomain "git.cin.su/m/Assistant/modules/capability/domain"
+)
+
+type fleetEndpoint struct {
+	ProviderID string
+	ModelID    string
+	Tier       capability.FleetTier
+	URL        string
+}
+
+type fleetStatusResponse struct {
+	Readiness capability.FleetReadiness `json:"readiness"`
+	Preferred capability.FleetTier      `json:"preferred_tier"`
+	Selected  capability.FleetTier      `json:"selected_tier,omitempty"`
+	Error     string                    `json:"error,omitempty"`
+}
+
+func fleetReadinessHandler(client *http.Client, preferred capability.FleetTier, endpoints []fleetEndpoint) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		members := make([]capability.FleetMember, 0, len(endpoints))
+		for _, endpoint := range endpoints {
+			member := capability.FleetMember{ProviderInstanceID: capability.ProviderInstanceID(endpoint.ProviderID), ModelDescriptorID: capability.ModelDescriptorID(endpoint.ModelID), Tier: endpoint.Tier, Reason: "NOT_CONFIGURED"}
+			if strings.TrimSpace(endpoint.URL) == "" {
+				members = append(members, member)
+				continue
+			}
+			request, err := http.NewRequestWithContext(r.Context(), http.MethodGet, endpoint.URL, nil)
+			if err == nil {
+				response, requestErr := client.Do(request)
+				if requestErr == nil {
+					response.Body.Close()
+					if response.StatusCode >= 200 && response.StatusCode < 300 {
+						member.Ready, member.Reason = true, "AVAILABLE"
+					} else {
+						member.Reason = fmt.Sprintf("status=%s", response.Status)
+					}
+				} else {
+					member.Reason = requestErr.Error()
+				}
+			}
+			members = append(members, member)
+		}
+		readiness, err := fleetdomain.BuildFleetReadiness(time.Now().UnixMilli(), members)
+		response := fleetStatusResponse{Readiness: readiness, Preferred: preferred}
+		if err == nil {
+			response.Selected, err = fleetdomain.SelectFleetTier(readiness, preferred)
+		}
+		if err != nil {
+			response.Error = err.Error()
+		}
+		w.Header().Set("Content-Type", "application/json")
+		if response.Selected == "" {
+			w.WriteHeader(http.StatusServiceUnavailable)
+		}
+		_ = json.NewEncoder(w).Encode(response)
+	})
+}
+
+func fleetContext(ctx context.Context) context.Context { return ctx }

cmd/cognitive-worker/fleet_status_test.go

@@ -0,0 +1,27 @@
+package main
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"git.cin.su/m/Assistant/contracts/capability"
+)
+
+func TestFleetReadinessSelectsLifeboatWhenPrimaryTiersUnavailable(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }))
+	defer server.Close()
+	handler := fleetReadinessHandler(server.Client(), capability.FleetTierHeavy, []fleetEndpoint{
+		{ProviderID: "cpu", ModelID: "cpu-model", Tier: capability.FleetTierFast, URL: server.URL},
+		{ProviderID: "medium", ModelID: "medium-model", Tier: capability.FleetTierMedium},
+		{ProviderID: "heavy", ModelID: "heavy-model", Tier: capability.FleetTierHeavy},
+		{ProviderID: "lifeboat", ModelID: "lifeboat-model", Tier: capability.FleetTierLifeboat, URL: server.URL},
+	})
+	recorder := httptest.NewRecorder()
+	request := httptest.NewRequest(http.MethodGet, "/v1/fleet/readiness", nil)
+	handler.ServeHTTP(recorder, request)
+	if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), `"selected_tier":"LIFEBOAT"`) || !strings.Contains(recorder.Body.String(), `"tier":"MEDIUM"`) {
+		t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
+	}
+}

cmd/cognitive-worker/main.go

@@ -76,10 +76,18 @@ func main() {
 	lifeboatURL := strings.TrimSuffix(strings.TrimRight(envOr("LOCAL_LIFEBOAT_PROVIDER_URL", "http://cognitive-lifeboat:8086/v1"), "/"), "/v1") + "/health"
 	mux.Handle("/readyz", health.ReadinessHandler{Runtime: "cognitive-worker", Probes: []health.Probe{
 		health.HTTPProbe(readinessClient, strings.TrimRight(envOr("CORE_RUNTIME_URL", "http://core-runtime:8080"), "/")+"/healthz"),
-		health.HTTPProbe(readinessClient, strings.TrimRight(envOr("LOCAL_CPU_PROVIDER_URL", "http://docker.cin.su:18082/v1"), "/")+"/models"),
+		health.OptionalHTTPProbe(readinessClient, strings.TrimRight(envOr("LOCAL_CPU_PROVIDER_URL", "http://docker.cin.su:18082/v1"), "/")+"/models"),
 		health.HTTPProbe(readinessClient, lifeboatURL),
-		health.HTTPProbe(readinessClient, strings.TrimRight(envOr("LOCAL_GPU_PROVIDER_URL", "http://192.168.220.91:11434"), "/")+"/api/tags"),
+		health.OptionalHTTPProbe(readinessClient, strings.TrimRight(envOr("LOCAL_GPU_PROVIDER_URL", "http://192.168.220.91:11434"), "/")+"/api/tags"),
 	}})
+	fleetClient := &http.Client{Timeout: 2 * time.Second}
+	preferredFleet := capability.FleetTier(envOr("COGNITIVE_PREFERRED_FLEET_TIER", string(capability.FleetTierHeavy)))
+	mux.Handle("/v1/fleet/readiness", fleetReadinessHandler(fleetClient, preferredFleet, []fleetEndpoint{
+		{ProviderID: "local-cpu-primary", ModelID: "local-cpu-model", Tier: capability.FleetTierFast, URL: strings.TrimRight(envOr("LOCAL_CPU_PROVIDER_URL", "http://docker.cin.su:18082/v1"), "/") + "/models"},
+		{ProviderID: "local-medium-unconfigured", ModelID: "medium-unconfigured", Tier: capability.FleetTierMedium},
+		{ProviderID: "local-gpu-opportunistic", ModelID: "gpu-opportunistic", Tier: capability.FleetTierHeavy, URL: strings.TrimRight(envOr("LOCAL_GPU_PROVIDER_URL", "http://192.168.220.91:11434"), "/") + "/api/tags"},
+		{ProviderID: "cognitive-lifeboat", ModelID: "qwen3-8b-lifeboat", Tier: capability.FleetTierLifeboat, URL: lifeboatURL},
+	}))
 	mux.HandleFunc("POST /v1/proposals/deterministic", deterministicProposal)
 	mux.HandleFunc("POST /v1/advisory-policy", gateway.policyHandler)
 	mux.HandleFunc("POST /v1/advisories", gateway.advisoryHandler)

cmd/research-runtime/main.go

@@ -0,0 +1,126 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+
+	"git.cin.su/m/Assistant/adapters/postgres"
+	researchadapter "git.cin.su/m/Assistant/adapters/research"
+	"git.cin.su/m/Assistant/contracts/epistemics"
+	secretcontracts "git.cin.su/m/Assistant/contracts/secrets"
+	"git.cin.su/m/Assistant/modules/epistemics/application"
+	"git.cin.su/m/Assistant/platform/secrets"
+	"git.cin.su/m/Assistant/runtime/health"
+)
+
+func main() {
+	ctx := context.Background()
+	password, err := resolveDatabasePassword()
+	if err != nil {
+		log.Fatal(err)
+	}
+	repository, err := postgres.OpenEpistemics(ctx, postgres.ContinuityConfig{Host: os.Getenv("POSTGRES_HOST"), Port: os.Getenv("POSTGRES_PORT"), Database: os.Getenv("POSTGRES_DATABASE"), User: os.Getenv("EPISTEMICS_POSTGRES_USER"), Password: password})
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer repository.Close()
+
+	mux := http.NewServeMux()
+	mux.Handle("/healthz", health.Handler{Runtime: "research-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1})
+	mux.Handle("/readyz", health.ReadinessHandler{Runtime: "research-runtime", Probes: []health.Probe{{Name: "epistemics-postgres", Check: func(ctx context.Context) error { return repository.Ping(ctx) }}}})
+	mux.HandleFunc("POST /v1/research/evaluate", func(w http.ResponseWriter, r *http.Request) {
+		var request epistemics.ResearchEvaluationRequest
+		if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 2<<20)).Decode(&request); err != nil {
+			http.Error(w, "invalid research evidence", http.StatusBadRequest)
+			return
+		}
+		result, err := application.EvaluateAndPersist(r.Context(), request, repository, time.Now().UnixMilli())
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusUnprocessableEntity)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(result)
+	})
+	connector := researchadapter.RepositoryConnector{Root: os.Getenv("RESEARCH_SOURCE_ROOT")}
+	mux.HandleFunc("POST /v1/research/repository/evaluate", func(w http.ResponseWriter, r *http.Request) {
+		var request epistemics.RepositoryResearchRequest
+		if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 128<<10)).Decode(&request); err != nil {
+			http.Error(w, "invalid repository research request", http.StatusBadRequest)
+			return
+		}
+		candidate, err := connector.Collect(r.Context(), request, time.Now().UnixMilli())
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusUnprocessableEntity)
+			return
+		}
+		result, err := application.EvaluateAndPersist(r.Context(), epistemics.ResearchEvaluationRequest{Plan: request.Plan, Candidates: []epistemics.ResearchCandidate{candidate}}, repository, time.Now().UnixMilli())
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusUnprocessableEntity)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(result)
+	})
+	webConnector := researchadapter.HTTPConnector{AllowedHosts: allowedHosts(os.Getenv("RESEARCH_HTTP_ALLOWED_HOSTS")), Client: &http.Client{Timeout: 5 * time.Second}}
+	mux.HandleFunc("POST /v1/research/web/evaluate", func(w http.ResponseWriter, r *http.Request) {
+		var request epistemics.WebResearchRequest
+		if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 128<<10)).Decode(&request); err != nil {
+			http.Error(w, "invalid web research request", http.StatusBadRequest)
+			return
+		}
+		candidate, err := webConnector.Collect(r.Context(), request, time.Now().UnixMilli())
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusUnprocessableEntity)
+			return
+		}
+		result, err := application.EvaluateAndPersist(r.Context(), epistemics.ResearchEvaluationRequest{Plan: request.Plan, Candidates: []epistemics.ResearchCandidate{candidate}}, repository, time.Now().UnixMilli())
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusUnprocessableEntity)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(result)
+	})
+	addr := os.Getenv("LISTEN_ADDR")
+	if addr == "" {
+		addr = ":8087"
+	}
+	log.Printf("research-runtime listening on %s", addr)
+	log.Fatal(http.ListenAndServe(addr, mux))
+}
+
+func allowedHosts(value string) map[string]struct{} {
+	result := map[string]struct{}{}
+	for _, host := range strings.Split(value, ",") {
+		if host = strings.ToLower(strings.TrimSpace(host)); host != "" {
+			result[host] = struct{}{}
+		}
+	}
+	return result
+}
+
+func resolveDatabasePassword() (string, error) {
+	if _, ok := os.LookupEnv("EPISTEMICS_POSTGRES_PASSWORD"); ok {
+		return "", fmt.Errorf("EPISTEMICS_POSTGRES_PASSWORD direct environment secret is forbidden")
+	}
+	path, ref := os.Getenv("EPISTEMICS_POSTGRES_PASSWORD_FILE"), os.Getenv("EPISTEMICS_POSTGRES_CREDENTIAL_REF")
+	if path == "" || ref == "" {
+		return "", fmt.Errorf("EPISTEMICS_POSTGRES_PASSWORD_FILE and EPISTEMICS_POSTGRES_CREDENTIAL_REF are required")
+	}
+	root := os.Getenv("SECRET_BACKEND_ROOT")
+	if root == "" {
+		root = "/run/secrets"
+	}
+	resolver, err := secrets.NewFileResolver(root)
+	if err != nil {
+		return "", err
+	}
+	return resolver.ResolveFile(secretcontracts.SecretFileRef{CredentialRef: secretcontracts.CredentialRef(ref), Purpose: secretcontracts.SecretPurposeDatabasePassword, Consumer: secretcontracts.RuntimeConsumerEpistemicsWorker, Path: path})
+}

cmd/work-graph-runtime/main.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+
+	"git.cin.su/m/Assistant/adapters/postgres"
+	secretcontracts "git.cin.su/m/Assistant/contracts/secrets"
+	"git.cin.su/m/Assistant/modules/work/application"
+	"git.cin.su/m/Assistant/platform/secrets"
+	"git.cin.su/m/Assistant/runtime/health"
+)
+
+func main() {
+	ctx := context.Background()
+	password, err := resolveDatabasePassword()
+	if err != nil {
+		log.Fatal(err)
+	}
+	repository, err := postgres.OpenWork(ctx, postgres.ContinuityConfig{Host: os.Getenv("POSTGRES_HOST"), Port: os.Getenv("POSTGRES_PORT"), Database: os.Getenv("POSTGRES_DATABASE"), User: os.Getenv("WORK_POSTGRES_USER"), Password: password})
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer repository.Close()
+	mux := http.NewServeMux()
+	mux.Handle("/healthz", health.Handler{Runtime: "work-graph-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1})
+	mux.Handle("/readyz", health.ReadinessHandler{Runtime: "work-graph-runtime", Probes: []health.Probe{{Name: "work-postgres", Check: func(ctx context.Context) error { return repository.Ping(ctx) }}}})
+	mux.HandleFunc("POST /v1/work-graphs/schedule", func(w http.ResponseWriter, r *http.Request) {
+		var request application.GraphScheduleRequest
+		if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&request); err != nil {
+			http.Error(w, "invalid graph schedule request", http.StatusBadRequest)
+			return
+		}
+		graph, ready, err := application.AdmitAndSchedule(r.Context(), repository, request)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusUnprocessableEntity)
+			return
+		}
+		writeJSON(w, map[string]any{"graph": graph, "scheduled_nodes": ready})
+	})
+	mux.HandleFunc("POST /v1/work-graphs/complete", func(w http.ResponseWriter, r *http.Request) {
+		var request application.GraphCompleteRequest
+		if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&request); err != nil {
+			http.Error(w, "invalid graph completion request", http.StatusBadRequest)
+			return
+		}
+		graph, err := application.CompleteGraphNode(r.Context(), repository, request.GraphID, request.Revision, request.NodeID, request.Succeeded)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusUnprocessableEntity)
+			return
+		}
+		writeJSON(w, graph)
+	})
+	addr := os.Getenv("LISTEN_ADDR")
+	if addr == "" {
+		addr = ":8088"
+	}
+	log.Printf("work-graph-runtime listening on %s", addr)
+	log.Fatal(http.ListenAndServe(addr, mux))
+}
+
+func writeJSON(w http.ResponseWriter, value any) {
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(value)
+}
+
+func resolveDatabasePassword() (string, error) {
+	if _, ok := os.LookupEnv("WORK_POSTGRES_PASSWORD"); ok {
+		return "", fmt.Errorf("WORK_POSTGRES_PASSWORD direct environment secret is forbidden")
+	}
+	path, ref := os.Getenv("WORK_POSTGRES_PASSWORD_FILE"), os.Getenv("WORK_POSTGRES_CREDENTIAL_REF")
+	if path == "" || ref == "" {
+		return "", fmt.Errorf("WORK_POSTGRES_PASSWORD_FILE and WORK_POSTGRES_CREDENTIAL_REF are required")
+	}
+	root := os.Getenv("SECRET_BACKEND_ROOT")
+	if root == "" {
+		root = "/run/secrets"
+	}
+	resolver, err := secrets.NewFileResolver(root)
+	if err != nil {
+		return "", err
+	}
+	return resolver.ResolveFile(secretcontracts.SecretFileRef{CredentialRef: secretcontracts.CredentialRef(ref), Purpose: secretcontracts.SecretPurposeDatabasePassword, Consumer: secretcontracts.RuntimeConsumerWorkWorker, Path: path})
+}

contracts/capability/fleet.go

@@ -10,16 +10,16 @@ const (
 )
 
 type FleetMember struct {
-	ProviderInstanceID ProviderInstanceID
-	ModelDescriptorID  ModelDescriptorID
-	Tier               FleetTier
-	Ready              bool
-	Reason             string
+	ProviderInstanceID ProviderInstanceID `json:"provider_instance_id"`
+	ModelDescriptorID  ModelDescriptorID  `json:"model_descriptor_id"`
+	Tier               FleetTier          `json:"tier"`
+	Ready              bool               `json:"ready"`
+	Reason             string             `json:"reason"`
 }
 
 type FleetReadiness struct {
-	ObservedAtUnixMillis int64
-	Members              []FleetMember
-	ReadyTiers           []FleetTier
-	FallbackTier         FleetTier
+	ObservedAtUnixMillis int64         `json:"observed_at_unix_millis"`
+	Members              []FleetMember `json:"members"`
+	ReadyTiers           []FleetTier   `json:"ready_tiers"`
+	FallbackTier         FleetTier     `json:"fallback_tier"`
 }

contracts/epistemics/research.go

@@ -106,6 +106,31 @@ type ResearchResult struct {
 	EvaluatedAtUnixMillis int64
 }
 
+// ResearchEvaluationRequest carries immutable evidence from a read-only
+// connector to the evaluator. The runtime never fetches or mutates a source
+// from this contract.
+type ResearchEvaluationRequest struct {
+	Plan       ResearchPlan
+	Candidates []ResearchCandidate
+}
+
+type RepositoryResearchRequest struct {
+	Plan           ResearchPlan
+	PropositionKey string
+	SourceID       SourceID
+	Path           string
+	Statement      string
+}
+
+type WebResearchRequest struct {
+	Plan           ResearchPlan
+	PropositionKey string
+	SourceID       SourceID
+	URL            string
+	Title          string
+	Statement      string
+}
+
 // ResolveCitation creates evidence only when all identifiers form one exact
 // source chain. It deliberately does not fetch, infer or rewrite source text.
 func ResolveCitation(source Source, snapshot SourceSnapshot, excerpt SourceExcerpt, evidence EvidenceKind, coverage CoverageState, conflict ConflictState, resolvedAtUnixMillis int64) (Citation, error) {

contracts/epistemics/truth.go

@@ -31,6 +31,9 @@ type Belief struct {
 }
 
 type TruthBundle struct {
+	Source      Source
+	Snapshot    SourceSnapshot
+	Excerpt     SourceExcerpt
 	Observation Observation
 	Claim       Claim
 	Belief      Belief

contracts/work/bridge.go

@@ -0,0 +1,32 @@
+package work
+
+import "time"
+
+// AgentExecutionRequest is the serializable hand-off from the durable worker
+// to an executor. The bridge receives the already-authorized dispatch context;
+// it does not receive database credentials or ambient authority.
+type AgentExecutionRequest struct {
+	Intent         AgentDispatchIntent `json:"intent"`
+	WorkerID       string              `json:"worker_id"`
+	FencingToken   AgentFencingToken   `json:"fencing_token"`
+	LeaseExpiresAt time.Time           `json:"lease_expires_at"`
+}
+
+type AgentExecutionState string
+
+const (
+	AgentExecutionCompleted AgentExecutionState = "COMPLETED"
+	AgentExecutionWaiting   AgentExecutionState = "WAITING"
+	AgentExecutionFailed    AgentExecutionState = "FAILED"
+)
+
+// AgentExecutionResult is an acknowledgement of the external execution
+// attempt. The worker persists it through the work owner before completing a
+// dispatch, so a bridge timeout never becomes an unrecorded success.
+type AgentExecutionResult struct {
+	State                AgentExecutionState `json:"state"`
+	Summary              string              `json:"summary"`
+	OutcomeRef           string              `json:"outcome_ref"`
+	ArtifactRefs         []string            `json:"artifact_refs"`
+	ObservedAtUnixMillis int64               `json:"observed_at_unix_millis"`
+}

contracts/work/dispatch_queue.go

@@ -10,10 +10,10 @@ const (
 )
 
 type AgentDispatchQueueItem struct {
-	DispatchID            string
-	State                 AgentDispatchQueueState
-	AvailableAtUnixMillis int64
-	Attempts              int
-	ClaimedBy             string
-	UpdatedAtUnixMillis   int64
+	DispatchID            string                  `json:"dispatch_id"`
+	State                 AgentDispatchQueueState `json:"state"`
+	AvailableAtUnixMillis int64                   `json:"available_at_unix_millis"`
+	Attempts              int                     `json:"attempts"`
+	ClaimedBy             string                  `json:"claimed_by"`
+	UpdatedAtUnixMillis   int64                   `json:"updated_at_unix_millis"`
 }

contracts/work/graph.go

@@ -26,29 +26,30 @@ const (
 )
 
 type GraphBudget struct {
-	MaxParallel int
-	MaxAttempts int
+	MaxParallel int `json:"max_parallel"`
+	MaxAttempts int `json:"max_attempts"`
 }
 
 type GraphNode struct {
-	ID             GraphNodeID
-	IdempotencyKey string
-	State          GraphNodeState
-	Attempts       int
+	ID             GraphNodeID    `json:"id"`
+	DispatchID     string         `json:"dispatch_id"`
+	IdempotencyKey string         `json:"idempotency_key"`
+	State          GraphNodeState `json:"state"`
+	Attempts       int            `json:"attempts"`
 }
 
 type GraphEdge struct {
-	From GraphNodeID
-	To   GraphNodeID
+	From GraphNodeID `json:"from"`
+	To   GraphNodeID `json:"to"`
 }
 
 type WorkGraph struct {
-	ID       string
-	Revision int64
-	State    GraphState
-	Nodes    []GraphNode
-	Edges    []GraphEdge
-	Budget   GraphBudget
+	ID       string      `json:"id"`
+	Revision int64       `json:"revision"`
+	State    GraphState  `json:"state"`
+	Nodes    []GraphNode `json:"nodes"`
+	Edges    []GraphEdge `json:"edges"`
+	Budget   GraphBudget `json:"budget"`
 }
 
 type ResultCandidate struct {

contracts/work/lease.go

@@ -50,14 +50,14 @@ type IndependentVerification struct {
 }
 
 type AgentDispatchIntent struct {
-	DispatchID             string
-	IdempotencyKey         string
-	DelegationID           DelegationID
-	RunID                  AgentRunID
-	ImplementationID       string
-	ImplementationRevision int64
-	SourceRevision         string
-	WorktreeID             string
-	VerificationPlanRef    string
-	CreatedAtUnixMillis    int64
+	DispatchID             string       `json:"dispatch_id"`
+	IdempotencyKey         string       `json:"idempotency_key"`
+	DelegationID           DelegationID `json:"delegation_id"`
+	RunID                  AgentRunID   `json:"run_id"`
+	ImplementationID       string       `json:"implementation_id"`
+	ImplementationRevision int64        `json:"implementation_revision"`
+	SourceRevision         string       `json:"source_revision"`
+	WorktreeID             string       `json:"worktree_id"`
+	VerificationPlanRef    string       `json:"verification_plan_ref"`
+	CreatedAtUnixMillis    int64        `json:"created_at_unix_millis"`
 }

database/migrations/034_epistemics_truth_bundles.sql

@@ -0,0 +1,49 @@
+BEGIN;
+
+SET ROLE epistemics_role;
+
+CREATE TABLE IF NOT EXISTS epistemics.truth_bundles (
+  bundle_id TEXT PRIMARY KEY,
+  source_id TEXT NOT NULL,
+  source_kind TEXT NOT NULL,
+  source_locator TEXT NOT NULL,
+  source_title TEXT NOT NULL,
+  source_publisher TEXT NOT NULL,
+  snapshot_id TEXT NOT NULL,
+  snapshot_source_id TEXT NOT NULL,
+  snapshot_content_digest TEXT NOT NULL,
+  snapshot_captured_at_unix_millis BIGINT NOT NULL,
+  snapshot_freshness TEXT NOT NULL,
+  excerpt_id TEXT NOT NULL,
+  excerpt_snapshot_id TEXT NOT NULL,
+  excerpt_text TEXT NOT NULL,
+  excerpt_start BIGINT NOT NULL,
+  excerpt_end BIGINT NOT NULL,
+  observation_id TEXT NOT NULL,
+  observation_observed_at_unix_millis BIGINT NOT NULL,
+  claim_id TEXT NOT NULL,
+  claim_statement TEXT NOT NULL,
+  claim_observation_id TEXT NOT NULL,
+  claim_citation_id TEXT NOT NULL,
+  claim_verification TEXT NOT NULL,
+  belief_id TEXT NOT NULL,
+  belief_claim_id TEXT NOT NULL,
+  belief_state TEXT NOT NULL,
+  citation_id TEXT NOT NULL,
+  citation_excerpt_id TEXT NOT NULL,
+  citation_evidence TEXT NOT NULL,
+  citation_coverage TEXT NOT NULL,
+  citation_conflict TEXT NOT NULL,
+  citation_resolved_at_unix_millis BIGINT NOT NULL,
+  recorded_at_unix_millis BIGINT NOT NULL,
+  UNIQUE (snapshot_id, excerpt_id, claim_id),
+  CHECK (bundle_id <> ''), CHECK (source_id <> ''), CHECK (source_locator <> ''),
+  CHECK (excerpt_start >= 0), CHECK (excerpt_end >= excerpt_start),
+  CHECK (recorded_at_unix_millis > 0)
+);
+
+CREATE INDEX IF NOT EXISTS truth_bundles_source_time_idx
+  ON epistemics.truth_bundles (source_id, recorded_at_unix_millis DESC);
+
+RESET ROLE;
+COMMIT;

database/migrations/035_work_graph_dispatch_binding.sql

@@ -0,0 +1,8 @@
+BEGIN;
+
+SET ROLE work_role;
+
+ALTER TABLE work.graph_nodes ADD COLUMN IF NOT EXISTS dispatch_id TEXT NOT NULL DEFAULT '';
+
+RESET ROLE;
+COMMIT;

deploy/docker-compose.prod.yml

@@ -118,6 +118,22 @@ services:
     volumes:
       - assistant_core_secrets:/run/secrets:ro
 
+  work-graph-runtime:
+    build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: work-graph-runtime, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } }
+    restart: unless-stopped
+    environment:
+      LISTEN_ADDR: :8088
+      POSTGRES_HOST: postgres
+      POSTGRES_PORT: "5432"
+      POSTGRES_DATABASE: ${POSTGRES_DB:-assistant}
+      WORK_POSTGRES_USER: work_role
+      WORK_POSTGRES_CREDENTIAL_REF: credential://domovoy/work/db-password/v1
+      WORK_POSTGRES_PASSWORD_FILE: /run/secrets/work_db_password
+      SECRET_BACKEND_ROOT: /run/secrets
+    depends_on: { postgres: { condition: service_healthy } }
+    volumes:
+      - assistant_core_secrets:/run/secrets:ro
+
   reflection-worker:
     build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: reflection-worker, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } }
     restart: unless-stopped
@@ -134,6 +150,25 @@ services:
     volumes:
       - assistant_core_secrets:/run/secrets:ro
 
+  research-runtime:
+    build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: research-runtime, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } }
+    restart: unless-stopped
+    environment:
+      LISTEN_ADDR: :8087
+      RESEARCH_SOURCE_ROOT: /var/lib/research-sources
+      RESEARCH_HTTP_ALLOWED_HOSTS: assistant.cin.su,git.cin.su
+      POSTGRES_HOST: postgres
+      POSTGRES_PORT: "5432"
+      POSTGRES_DATABASE: ${POSTGRES_DB:-assistant}
+      EPISTEMICS_POSTGRES_USER: epistemics_role
+      EPISTEMICS_POSTGRES_CREDENTIAL_REF: credential://domovoy/epistemics/db-password/v1
+      EPISTEMICS_POSTGRES_PASSWORD_FILE: /run/secrets/epistemics_db_password
+      SECRET_BACKEND_ROOT: /run/secrets
+    depends_on: { postgres: { condition: service_healthy } }
+    volumes:
+      - assistant_core_secrets:/run/secrets:ro
+      - assistant_research_sources:/var/lib/research-sources:ro
+
   continuity-runtime:
     build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: continuity-runtime, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } }
     restart: unless-stopped
@@ -242,5 +277,7 @@ volumes:
   assistant_notification_reader_secret:
     external: true
   llm_local_models:
-    external: true
     name: llm-local-models
+    external: true
+  assistant_research_sources:
+    name: assistant_research_sources

dev/ACTIVE_TASK.json

@@ -1,5 +1,5 @@
 {
-  "task_file": "dev/tasks/DEV-000115-cognitive-admission-control.json",
-  "task_id": "DEV-000115",
-  "task_revision": 6
+  "task_file": "dev/tasks/DEV-000146-work-graph-runtime.json",
+  "task_id": "DEV-000146",
+  "task_revision": 1
 }

dev/tasks/DEV-000141-optional-fleet-readiness.json

@@ -0,0 +1,23 @@
+{
+  "task_id": "DEV-000141",
+  "task_revision": 2,
+  "architecture_revision": "ADR-0018",
+  "base_repo_revision": "38cff16",
+  "risk_tier": "E",
+  "intent": "Keep cognitive readiness available when an optional opportunistic fleet member is unavailable while preserving typed degraded evidence.",
+  "requirement_ids": ["WP-003", "WP-203", "Fleet failover readiness"],
+  "allowed_change_scope": {"modules": ["runtime-health", "cognitive-worker"], "paths": ["runtime/health/**", "cmd/cognitive-worker/main.go", "runtime/health/server_test.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000141-optional-fleet-readiness.json"]},
+  "forbidden_change_scope": {"paths": ["architecture/**", "database/**", "deploy/**", "security/**", ".gitea/**"]},
+  "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 0},
+  "contracts": {"consume": ["Existing readiness Probe and typed readiness response"], "may_change": ["Optional dependency probe behavior"]},
+  "semantic_owner": "runtime-health and cognitive-worker",
+  "canonical_state": [],
+  "state_transitions": ["optional provider unavailable -> readiness remains ready with typed degraded check", "required dependency unavailable -> readiness remains degraded"],
+  "side_effects_and_privileges": ["No external mutation; no provider invocation"],
+  "persistence_and_migrations": ["None"],
+  "failure_retry_restart": ["Optional provider can recover on subsequent readiness probe"],
+  "assumptions": ["CPU/Lifeboat remains a qualified fallback when opportunistic GPU is unavailable"],
+  "acceptance_scenarios": ["GPU timeout does not hide degraded evidence or make fallback unavailable", "required dependency failure still returns 503"],
+  "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "docker compose config"],
+  "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"]
+}

dev/tasks/DEV-000142-truth-bundle-persistence.json

@@ -0,0 +1,23 @@
+{
+  "task_id": "DEV-000142",
+  "task_revision": 1,
+  "architecture_revision": "ADR-0018",
+  "base_repo_revision": "0b321fe",
+  "risk_tier": "E",
+  "intent": "Persist typed Truth/Research bundles with exact source, snapshot, excerpt and citation provenance in the epistemics owner schema.",
+  "requirement_ids": ["WP-102", "Truth runtime", "Research provenance persistence"],
+  "allowed_change_scope": {"modules": ["epistemics", "database"], "paths": ["contracts/epistemics/truth.go", "modules/epistemics/domain/**", "adapters/postgres/epistemics_repository.go", "database/migrations/034_epistemics_truth_bundles.sql", "tests/research_contracts/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000142-truth-bundle-persistence.json"]},
+  "forbidden_change_scope": {"paths": ["architecture/**", "deploy/**", "security/**", "cmd/**", ".gitea/**"]},
+  "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 1, "new_third_party_dependencies": 0, "new_database_owner_roles": 1},
+  "contracts": {"consume": ["TruthBundle, Source, SourceSnapshot, SourceExcerpt and Citation contracts"], "may_change": ["TruthBundle carries exact source evidence needed for durable provenance"]},
+  "semantic_owner": "epistemics",
+  "canonical_state": ["epistemics.truth_bundles"],
+  "state_transitions": ["compiled truth bundle -> durable append/update by exact citation identity"],
+  "side_effects_and_privileges": ["epistemics_role writes only epistemics.truth_bundles"],
+  "persistence_and_migrations": ["Migration 034; idempotent table/index creation"],
+  "failure_retry_restart": ["bundle write is idempotent by citation-derived bundle id"],
+  "assumptions": ["external connectors remain read-only and are a separate follow-up"],
+  "acceptance_scenarios": ["exact source chain is persisted", "duplicate bundle write is idempotent", "cross-owner schemas are untouched"],
+  "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "docker compose config"],
+  "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT", "REQUIRES_MIGRATION_DECISION"]
+}

dev/tasks/DEV-000143-research-runtime-orchestration.json

@@ -0,0 +1,23 @@
+{
+  "task_id": "DEV-000143",
+  "task_revision": 3,
+  "architecture_revision": "ADR-0018",
+  "base_repo_revision": "f9e2173",
+  "risk_tier": "E",
+  "intent": "Run a private research runtime that evaluates immutable connector evidence and durably persists exact TruthBundle provenance.",
+  "requirement_ids": ["WP-102", "Truth runtime", "Research persistence orchestration"],
+  "allowed_change_scope": {"modules": ["epistemics", "research-runtime", "deployment"], "paths": ["contracts/epistemics/research.go", "modules/epistemics/application/**", "adapters/postgres/epistemics_repository.go", "adapters/research/**", "cmd/research-runtime/**", "deploy/docker-compose.prod.yml", "tests/research_runtime/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000143-research-runtime-orchestration.json"]},
+  "forbidden_change_scope": {"paths": ["architecture/**", "database/**", "security/**", ".gitea/**"]},
+  "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 1, "new_third_party_dependencies": 0, "new_database_owner_roles": 0},
+  "contracts": {"consume": ["ResearchEvaluationRequest, ResearchResult, TruthBundle and epistemics repository"], "may_change": ["Private persistence orchestration endpoint"]},
+  "semantic_owner": "epistemics",
+  "canonical_state": ["epistemics.truth_bundles"],
+  "state_transitions": ["repository/web source -> immutable digest/excerpt -> evaluated finding -> persisted provenance"],
+  "side_effects_and_privileges": ["epistemics_role writes through repository only; no source mutation or provider call"],
+  "persistence_and_migrations": ["Consumes migration 034 already in main"],
+  "failure_retry_restart": ["idempotent repository writes; runtime restart-safe"],
+  "assumptions": ["source fetching connectors are read-only callers and remain a separate package"],
+  "acceptance_scenarios": ["runtime ready with epistemics DB", "valid evidence is persisted", "invalid evidence is rejected", "no public compose port is exposed"],
+  "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "docker compose config"],
+  "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT", "REQUIRES_MIGRATION_DECISION"]
+}

dev/tasks/DEV-000144-fleet-readiness-runtime.json

@@ -0,0 +1,23 @@
+{
+  "task_id": "DEV-000144",
+  "task_revision": 1,
+  "architecture_revision": "ADR-0018",
+  "base_repo_revision": "23b04aa",
+  "risk_tier": "E",
+  "intent": "Expose explainable runtime fleet readiness and explicit fallback tier from live provider probes.",
+  "requirement_ids": ["WP-203", "WP-601", "Model fleet/failover"],
+  "allowed_change_scope": {"modules": ["capability", "cognitive-worker"], "paths": ["contracts/capability/fleet.go", "cmd/cognitive-worker/**", "tests/cognitive_registry/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000144-fleet-readiness-runtime.json"]},
+  "forbidden_change_scope": {"paths": ["architecture/**", "database/**", "deploy/**", "security/**", ".gitea/**"]},
+  "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 0},
+  "contracts": {"consume": ["FleetReadiness and SelectFleetTier"], "may_change": ["JSON projection and live probe endpoint"]},
+  "semantic_owner": "capability and cognitive-worker",
+  "canonical_state": [],
+  "state_transitions": ["live provider probes -> typed member states -> preferred tier or explicit Lifeboat fallback"],
+  "side_effects_and_privileges": ["Read-only HTTP probes only"],
+  "persistence_and_migrations": ["None"],
+  "failure_retry_restart": ["Each request re-probes providers; no in-memory health authority"],
+  "assumptions": ["Lifeboat is the explicit fallback tier"],
+  "acceptance_scenarios": ["unavailable CPU/GPU selects Lifeboat", "unconfigured Medium is visible", "no fallback returns typed 503"],
+  "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "docker compose config"],
+  "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"]
+}

dev/tasks/DEV-000145-agent-execution-bridge-runtime.json

@@ -0,0 +1,23 @@
+{
+  "task_id": "DEV-000145",
+  "task_revision": 1,
+  "architecture_revision": "ADR-0018",
+  "base_repo_revision": "7b06cd64a279db5aec67ada5f3c705f114ffd86b",
+  "risk_tier": "E",
+  "intent": "Connect the durable agent worker to an explicit typed external execution bridge with lease and callback evidence.",
+  "requirement_ids": ["WP-301", "Agent Execution runtime and recovery"],
+  "allowed_change_scope": {"modules": ["work", "agent-worker", "agentbridge"], "paths": ["contracts/work/bridge.go", "adapters/agentbridge/**", "modules/work/application/**", "modules/work/ports/**", "cmd/agent-worker/**", "tests/agent_execution/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000145-agent-execution-bridge-runtime.json"]},
+  "forbidden_change_scope": {"paths": ["architecture/**", "database/**", "deploy/**", "security/**", ".gitea/**", "contracts/authority/**"]},
+  "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 0},
+  "contracts": {"consume": ["AgentDispatchIntent", "AgentLease", "AgentCallback"], "may_change": ["serializable bridge request/result contract"]},
+  "semantic_owner": "work",
+  "canonical_state": ["work agent lease and callback head"],
+  "state_transitions": ["durable dispatch -> lease/heartbeat -> bridge result -> result callback -> completed or explicit waiting/failure"],
+  "side_effects_and_privileges": ["outbound HTTP call to configured execution bridge only; no credentials forwarded"],
+  "persistence_and_migrations": ["reuse existing work lease/callback persistence; no migration"],
+  "failure_retry_restart": ["bridge timeout is failure of the current attempt; durable dispatch remains evidence-bearing and stale claims are recoverable"],
+  "assumptions": ["an independently operated bridge will implement POST /v1/execute", "bridge URL is deployment configuration and absent URL preserves legacy dry-run until bridge capability is deployed"],
+  "acceptance_scenarios": ["typed bridge request reaches executor", "completed bridge result persists callback evidence", "invalid/unavailable bridge fails closed", "existing stale-claim recovery remains intact"],
+  "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "docker compose config"],
+  "terminal_gap_states": ["BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"]
+}

dev/tasks/DEV-000146-work-graph-runtime.json

@@ -0,0 +1,23 @@
+{
+  "task_id": "DEV-000146",
+  "task_revision": 1,
+  "architecture_revision": "ADR-0018",
+  "base_repo_revision": "16d714238e6e77a105731311eaa54f3f486d3431",
+  "risk_tier": "E",
+  "intent": "Provide a durable work-graph runtime with bounded fan-out scheduling, dispatch queue binding, and revisioned fan-in completion.",
+  "requirement_ids": ["WP-400", "Multi-Agent Work Graph"],
+  "allowed_change_scope": {"modules": ["work", "work-graph-runtime"], "paths": ["contracts/work/graph.go", "contracts/work/lease.go", "contracts/work/dispatch_queue.go", "modules/work/application/**", "modules/work/ports/**", "adapters/postgres/work_repository.go", "database/migrations/035_work_graph_dispatch_binding.sql", "cmd/work-graph-runtime/**", "deploy/docker-compose.prod.yml", "tests/agent_execution/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000146-work-graph-runtime.json"]},
+  "forbidden_change_scope": {"paths": ["architecture/**", "security/**", ".gitea/**", "contracts/authority/**", "modules/identity/**"]},
+  "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 1, "new_third_party_dependencies": 0, "new_database_owner_roles": 1},
+  "contracts": {"consume": ["WorkGraph", "AgentDispatchIntent", "AgentDispatchQueueItem"], "may_change": ["GraphNode dispatch binding serialization"]},
+  "semantic_owner": "work",
+  "canonical_state": ["work graph revision, graph nodes, graph edges, dispatch intent and queue"],
+  "state_transitions": ["DRAFT -> ADMITTED -> RUNNING -> PARTIAL/COMPLETED or FAILED", "BLOCKED_PREREQUISITE -> LEASED -> SUCCEEDED/FAILED", "each transition writes a new graph revision"],
+  "side_effects_and_privileges": ["work-role local graph and dispatch writes only"],
+  "persistence_and_migrations": ["035 adds work.graph_nodes.dispatch_id; migration runner applies it before runtime start"],
+  "failure_retry_restart": ["revisioned graph history survives process restart; dispatch queue claims remain recoverable by agent-worker"],
+  "assumptions": ["dispatch intents supplied with schedule request are already authorized by work-owned planning flow"],
+  "acceptance_scenarios": ["two independent nodes fan out under MaxParallel", "dependent join remains blocked until both predecessors succeed", "join completion yields graph COMPLETED", "cycle and missing dispatch bindings fail closed"],
+  "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "docker compose config"],
+  "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT", "REQUIRES_MIGRATION_DECISION"]
+}

modules/epistemics/application/persisted_research.go

@@ -0,0 +1,40 @@
+package application
+
+import (
+	"context"
+
+	"git.cin.su/m/Assistant/contracts/epistemics"
+	domain "git.cin.su/m/Assistant/modules/epistemics/domain"
+)
+
+type TruthBundleStore interface {
+	SaveTruthBundle(context.Context, epistemics.TruthBundle, int64) error
+}
+
+// EvaluateAndPersist evaluates only supplied immutable connector evidence and
+// persists the exact source chain used by each candidate. It never performs a
+// network fetch and never promotes a belief to verified state.
+func EvaluateAndPersist(ctx context.Context, request epistemics.ResearchEvaluationRequest, store TruthBundleStore, evaluatedAtUnixMillis int64) (epistemics.ResearchResult, error) {
+	result, err := (ResearchService{Connector: staticConnector{candidates: request.Candidates}}).Evaluate(ctx, request.Plan, evaluatedAtUnixMillis)
+	if err != nil {
+		return epistemics.ResearchResult{}, err
+	}
+	for _, candidate := range request.Candidates {
+		bundle, err := domain.CompileObservedClaim(candidate.Source, candidate.Snapshot, candidate.Excerpt, candidate.Statement, candidate.Snapshot.CapturedAtUnixMillis, candidate.Conflict)
+		if err != nil {
+			return epistemics.ResearchResult{}, err
+		}
+		if err := store.SaveTruthBundle(ctx, bundle, evaluatedAtUnixMillis); err != nil {
+			return epistemics.ResearchResult{}, err
+		}
+	}
+	return result, nil
+}
+
+type staticConnector struct {
+	candidates []epistemics.ResearchCandidate
+}
+
+func (c staticConnector) Collect(context.Context, epistemics.ResearchPlan) ([]epistemics.ResearchCandidate, error) {
+	return c.candidates, nil
+}

modules/epistemics/domain/research.go

@@ -106,5 +106,5 @@ func compileResearchCandidate(candidate epistemics.ResearchCandidate) (epistemic
 	}
 	observation := epistemics.Observation{ID: epistemics.ObservationID("observation:" + string(candidate.Excerpt.ID)), SourceSnapshot: candidate.Snapshot.ID, Excerpt: candidate.Excerpt.ID, Text: candidate.Excerpt.Text, ObservedAt: candidate.Snapshot.CapturedAtUnixMillis}
 	claim := epistemics.Claim{ID: epistemics.ClaimID("claim:" + string(candidate.Excerpt.ID)), Statement: candidate.Statement, ObservationID: observation.ID, CitationID: citation.ID, Verification: epistemics.VerificationUnverified}
-	return epistemics.TruthBundle{Observation: observation, Claim: claim, Belief: epistemics.Belief{ID: epistemics.BeliefID("belief:" + string(candidate.Excerpt.ID)), ClaimID: claim.ID, State: epistemics.BeliefProvisional}, Citation: citation}, nil
+	return epistemics.TruthBundle{Source: candidate.Source, Snapshot: candidate.Snapshot, Excerpt: candidate.Excerpt, Observation: observation, Claim: claim, Belief: epistemics.Belief{ID: epistemics.BeliefID("belief:" + string(candidate.Excerpt.ID)), ClaimID: claim.ID, State: epistemics.BeliefProvisional}, Citation: citation}, nil
 }

modules/epistemics/domain/truth.go

@@ -23,7 +23,7 @@ func CompileObservedClaim(source epistemics.Source, snapshot epistemics.SourceSn
 	if conflict == epistemics.ConflictPresent || conflict == epistemics.ConflictUnknown {
 		state = epistemics.BeliefConflicted
 	}
-	bundle := epistemics.TruthBundle{Observation: observation, Claim: claim, Belief: epistemics.Belief{ID: epistemics.BeliefID("belief:" + string(excerpt.ID)), ClaimID: claim.ID, State: state}, Citation: citation}
+	bundle := epistemics.TruthBundle{Source: source, Snapshot: snapshot, Excerpt: excerpt, Observation: observation, Claim: claim, Belief: epistemics.Belief{ID: epistemics.BeliefID("belief:" + string(excerpt.ID)), ClaimID: claim.ID, State: state}, Citation: citation}
 	if bundle.Observation.SourceSnapshot != snapshot.ID || bundle.Claim.ObservationID != bundle.Observation.ID || bundle.Claim.CitationID != bundle.Citation.ID || bundle.Belief.ClaimID != bundle.Claim.ID {
 		return epistemics.TruthBundle{}, ErrInvalidTruthBundle
 	}

modules/work/application/agent_dispatch.go

@@ -0,0 +1,52 @@
+package application
+
+import (
+	"context"
+	"time"
+
+	"git.cin.su/m/Assistant/contracts/work"
+	"git.cin.su/m/Assistant/modules/work/domain"
+	"git.cin.su/m/Assistant/modules/work/ports"
+)
+
+func ExecuteAgentDispatch(ctx context.Context, store ports.ExecutionStore, executor ports.AgentExecutor, intent work.AgentDispatchIntent, workerID string, token work.AgentFencingToken, now time.Time, ttl time.Duration) (work.AgentExecutionResult, error) {
+	if executor == nil {
+		return work.AgentExecutionResult{}, ports.ErrAgentExecutorUnavailable
+	}
+	if err := domain.ValidateDispatchIntent(intent); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	lease, err := domain.AcquireAgentLease(intent.RunID, workerID, token, now, ttl)
+	if err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	if err := store.SaveDispatchIntent(ctx, intent); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	if err := store.SaveAgentLease(ctx, lease); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	heartbeat := work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: 1, Kind: work.AgentCallbackHeartbeat, ObservedAt: now}
+	if err := domain.ValidateAgentCallback(heartbeat, lease, 0); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	if err := store.AcceptAgentCallback(ctx, heartbeat); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	result, err := executor.Execute(ctx, work.AgentExecutionRequest{Intent: intent, WorkerID: workerID, FencingToken: token, LeaseExpiresAt: lease.ExpiresAt})
+	if err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	observedAt := time.UnixMilli(result.ObservedAtUnixMillis)
+	callback := work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: 2, Kind: work.AgentCallbackResult, ObservedAt: observedAt}
+	if result.State == work.AgentExecutionWaiting {
+		callback.Kind = work.AgentCallbackCheckpoint
+	}
+	if err := domain.ValidateAgentCallback(callback, lease, heartbeat.Sequence); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	if err := store.AcceptAgentCallback(ctx, callback); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	return result, nil
+}

modules/work/application/agent_dispatch_test.go

@@ -0,0 +1,42 @@
+package application
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"git.cin.su/m/Assistant/contracts/work"
+)
+
+type bridgeStore struct {
+	callbacks []work.AgentCallback
+	lease     work.AgentLease
+}
+
+func (s *bridgeStore) SaveDispatchIntent(context.Context, work.AgentDispatchIntent) error { return nil }
+func (s *bridgeStore) SaveAgentLease(_ context.Context, lease work.AgentLease) error {
+	s.lease = lease
+	return nil
+}
+func (s *bridgeStore) AcceptAgentCallback(_ context.Context, callback work.AgentCallback) error {
+	s.callbacks = append(s.callbacks, callback)
+	return nil
+}
+
+type bridgeExecutor struct{}
+
+func (bridgeExecutor) Execute(context.Context, work.AgentExecutionRequest) (work.AgentExecutionResult, error) {
+	return work.AgentExecutionResult{State: work.AgentExecutionCompleted, Summary: "done", ObservedAtUnixMillis: 2}, nil
+}
+
+func TestExecuteAgentDispatchPersistsLeaseAndBridgeCallbacks(t *testing.T) {
+	store := &bridgeStore{}
+	now := time.UnixMilli(1)
+	result, err := ExecuteAgentDispatch(context.Background(), store, bridgeExecutor{}, work.AgentDispatchIntent{DispatchID: "d", IdempotencyKey: "i", DelegationID: "del", RunID: "run", ImplementationID: "impl", ImplementationRevision: 1, SourceRevision: "source", WorktreeID: "tree", VerificationPlanRef: "verify", CreatedAtUnixMillis: 1}, "worker", 1, now, time.Minute)
+	if err != nil || result.State != work.AgentExecutionCompleted {
+		t.Fatalf("result=%+v err=%v", result, err)
+	}
+	if len(store.callbacks) != 2 || store.callbacks[0].Kind != work.AgentCallbackHeartbeat || store.callbacks[1].Kind != work.AgentCallbackResult {
+		t.Fatalf("callbacks=%+v", store.callbacks)
+	}
+}

modules/work/application/graph_runtime.go

@@ -0,0 +1,143 @@
+package application
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"git.cin.su/m/Assistant/contracts/work"
+	"git.cin.su/m/Assistant/modules/work/domain"
+	"git.cin.su/m/Assistant/modules/work/ports"
+)
+
+var ErrGraphDispatchBinding = errors.New("graph node dispatch binding is missing")
+
+type GraphScheduleRequest struct {
+	Graph   work.WorkGraph             `json:"graph"`
+	Intents []work.AgentDispatchIntent `json:"intents"`
+	Now     int64                      `json:"now_unix_millis"`
+}
+
+type GraphCompleteRequest struct {
+	GraphID   string           `json:"graph_id"`
+	Revision  int64            `json:"revision"`
+	NodeID    work.GraphNodeID `json:"node_id"`
+	Succeeded bool             `json:"succeeded"`
+}
+
+func AdmitAndSchedule(ctx context.Context, store ports.GraphStore, request GraphScheduleRequest) (work.WorkGraph, []work.GraphNodeID, error) {
+	if request.Now < 1 {
+		request.Now = time.Now().UnixMilli()
+	}
+	admitted, err := domain.AdmitGraph(request.Graph)
+	if err != nil {
+		return work.WorkGraph{}, nil, err
+	}
+	if err := store.SaveWorkGraph(ctx, admitted); err != nil {
+		return work.WorkGraph{}, nil, err
+	}
+	ready, err := domain.ScheduleReadyNodes(admitted, 0)
+	if err != nil {
+		return work.WorkGraph{}, nil, err
+	}
+	intentByDispatch := make(map[string]work.AgentDispatchIntent, len(request.Intents))
+	for _, intent := range request.Intents {
+		intentByDispatch[intent.DispatchID] = intent
+	}
+	for _, node := range admitted.Nodes {
+		if node.DispatchID == "" {
+			return work.WorkGraph{}, nil, ErrGraphDispatchBinding
+		}
+		if _, ok := intentByDispatch[node.DispatchID]; !ok {
+			return work.WorkGraph{}, nil, ErrGraphDispatchBinding
+		}
+	}
+	for _, intent := range request.Intents {
+		if err := store.SaveDispatchIntent(ctx, intent); err != nil {
+			return work.WorkGraph{}, nil, err
+		}
+	}
+	next := admitted
+	next.Revision++
+	next.State = work.GraphRunning
+	selected := make(map[work.GraphNodeID]bool, len(ready))
+	for _, nodeID := range ready {
+		selected[nodeID] = true
+	}
+	for i := range next.Nodes {
+		if !selected[next.Nodes[i].ID] {
+			continue
+		}
+		if next.Nodes[i].DispatchID == "" {
+			return work.WorkGraph{}, nil, ErrGraphDispatchBinding
+		}
+		intent := intentByDispatch[next.Nodes[i].DispatchID]
+		if err := store.EnqueueDispatch(ctx, work.AgentDispatchQueueItem{DispatchID: intent.DispatchID, State: work.AgentDispatchQueued, AvailableAtUnixMillis: request.Now, UpdatedAtUnixMillis: request.Now}); err != nil {
+			return work.WorkGraph{}, nil, err
+		}
+		next.Nodes[i].State = work.GraphNodeLeased
+	}
+	if err := store.SaveWorkGraph(ctx, next); err != nil {
+		return work.WorkGraph{}, nil, err
+	}
+	return next, ready, nil
+}
+
+func CompleteGraphNode(ctx context.Context, store ports.GraphStore, graphID string, revision int64, nodeID work.GraphNodeID, success bool) (work.WorkGraph, error) {
+	graph, err := store.LoadWorkGraph(ctx, graphID, revision)
+	if err != nil {
+		return work.WorkGraph{}, err
+	}
+	found := false
+	for i := range graph.Nodes {
+		if graph.Nodes[i].ID == nodeID {
+			found = true
+			if success {
+				graph.Nodes[i].State = work.GraphNodeSuccess
+			} else {
+				graph.Nodes[i].State = work.GraphNodeFailed
+			}
+		}
+	}
+	if !found {
+		return work.WorkGraph{}, domain.ErrInvalidGraph
+	}
+	if !success {
+		graph.State = work.GraphFailed
+	} else {
+		ready, err := domain.ReadyGraphNodes(graph)
+		if err != nil {
+			return work.WorkGraph{}, err
+		}
+		for i := range graph.Nodes {
+			for _, candidate := range ready {
+				if graph.Nodes[i].ID == candidate && graph.Nodes[i].State == work.GraphNodeBlocked {
+					graph.Nodes[i].State = work.GraphNodeReady
+					if graph.Nodes[i].DispatchID == "" {
+						return work.WorkGraph{}, ErrGraphDispatchBinding
+					}
+					if err := store.EnqueueDispatch(ctx, work.AgentDispatchQueueItem{DispatchID: graph.Nodes[i].DispatchID, State: work.AgentDispatchQueued, AvailableAtUnixMillis: time.Now().UnixMilli(), UpdatedAtUnixMillis: time.Now().UnixMilli()}); err != nil {
+						return work.WorkGraph{}, err
+					}
+				}
+			}
+		}
+		allSuccess := true
+		for _, node := range graph.Nodes {
+			if node.State != work.GraphNodeSuccess {
+				allSuccess = false
+				break
+			}
+		}
+		if allSuccess {
+			graph.State = work.GraphCompleted
+		} else {
+			graph.State = work.GraphPartial
+		}
+	}
+	graph.Revision++
+	if err := store.SaveWorkGraph(ctx, graph); err != nil {
+		return work.WorkGraph{}, err
+	}
+	return graph, nil
+}

modules/work/application/worker.go

@@ -41,3 +41,34 @@ func ExecuteWorkerCycle(ctx context.Context, store ports.WorkerStore, workerID s
 	}
 	return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID, Completed: true}, nil
 }
+
+func ExecuteWorkerCycleWithExecutor(ctx context.Context, store ports.WorkerStore, executor ports.AgentExecutor, workerID string, token work.AgentFencingToken, now time.Time, staleAfter, leaseTTL time.Duration) (WorkerCycleResult, error) {
+	recovered, err := store.RecoverStaleClaims(ctx, now.UnixMilli(), staleAfter.Milliseconds())
+	if err != nil {
+		return WorkerCycleResult{}, err
+	}
+	item, err := store.ClaimNextDispatch(ctx, workerID, now.UnixMilli())
+	if err != nil {
+		return WorkerCycleResult{RecoveredClaims: recovered}, err
+	}
+	if item.DispatchID == "" {
+		return WorkerCycleResult{RecoveredClaims: recovered, Idle: true}, nil
+	}
+	intent, err := store.LoadDispatchIntent(ctx, item.DispatchID)
+	if err != nil {
+		_ = store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchFailed, now.UnixMilli())
+		return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err
+	}
+	result, err := ExecuteAgentDispatch(ctx, store, executor, intent, workerID, token, now, leaseTTL)
+	if err != nil {
+		_ = store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchFailed, now.UnixMilli())
+		return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err
+	}
+	if result.State != work.AgentExecutionCompleted {
+		return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, nil
+	}
+	if err := store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchCompleted, now.UnixMilli()); err != nil {
+		return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err
+	}
+	return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID, Completed: true}, nil
+}

modules/work/ports/execution.go

@@ -6,6 +6,10 @@ import (
 	"git.cin.su/m/Assistant/contracts/work"
 )
 
+type AgentExecutor interface {
+	Execute(context.Context, work.AgentExecutionRequest) (work.AgentExecutionResult, error)
+}
+
 type ExecutionStore interface {
 	SaveDispatchIntent(context.Context, work.AgentDispatchIntent) error
 	SaveAgentLease(context.Context, work.AgentLease) error

modules/work/ports/graph.go

@@ -0,0 +1,14 @@
+package ports
+
+import (
+	"context"
+
+	"git.cin.su/m/Assistant/contracts/work"
+)
+
+type GraphStore interface {
+	SaveWorkGraph(context.Context, work.WorkGraph) error
+	LoadWorkGraph(context.Context, string, int64) (work.WorkGraph, error)
+	SaveDispatchIntent(context.Context, work.AgentDispatchIntent) error
+	EnqueueDispatch(context.Context, work.AgentDispatchQueueItem) error
+}

modules/work/ports/worker.go

@@ -2,10 +2,13 @@ package ports
 
 import (
 	"context"
+	"errors"
 
 	"git.cin.su/m/Assistant/contracts/work"
 )
 
+var ErrAgentExecutorUnavailable = errors.New("agent executor unavailable")
+
 type WorkerStore interface {
 	ExecutionStore
 	LoadDispatchIntent(context.Context, string) (work.AgentDispatchIntent, error)

runtime/health/server.go

@@ -26,8 +26,9 @@ func (h Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 }
 
 type Probe struct {
-	Name  string
-	Check func(context.Context) error
+	Name     string
+	Check    func(context.Context) error
+	Optional bool
 }
 
 type Result struct {
@@ -59,11 +60,13 @@ func (h ReadinessHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			err := probe.Check(ctx)
 			cancel()
 			if err != nil {
-				ready = false
 				result.State, result.Reason, result.Detail = "DEGRADED", "DEPENDENCY_UNAVAILABLE", err.Error()
+				if !probe.Optional {
+					ready = false
+				}
 			}
 		}
-		if result.State != "READY" {
+		if result.State != "READY" && !probe.Optional {
 			ready = false
 		}
 		results = append(results, result)
@@ -96,6 +99,15 @@ func HTTPProbe(client *http.Client, endpoint string) Probe {
 	}}
 }
 
+// OptionalHTTPProbe reports dependency degradation without making the whole
+// runtime unavailable. This is used for opportunistic fleet members when a
+// qualified local fallback remains available.
+func OptionalHTTPProbe(client *http.Client, endpoint string) Probe {
+	probe := HTTPProbe(client, endpoint)
+	probe.Optional = true
+	return probe
+}
+
 func TCPProbe(name, host, port string) Probe {
 	return Probe{Name: name, Check: func(ctx context.Context) error {
 		connection, err := (&net.Dialer{}).DialContext(ctx, "tcp", net.JoinHostPort(host, port))

runtime/health/server_test.go

@@ -26,3 +26,16 @@ func TestReadinessReturns200WhenAllProbesPass(t *testing.T) {
 		t.Fatalf("readiness response = %d %s", recorder.Code, recorder.Body.String())
 	}
 }
+
+func TestReadinessStaysAvailableWithOptionalDependencyDegraded(t *testing.T) {
+	handler := ReadinessHandler{Runtime: "test", Probes: []Probe{
+		{Name: "lifeboat", Check: func(context.Context) error { return nil }},
+		{Name: "opportunistic-gpu", Optional: true, Check: func(context.Context) error { return errors.New("timeout") }},
+	}}
+	recorder := httptest.NewRecorder()
+	handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/readyz", nil))
+	body := recorder.Body.String()
+	if recorder.Code != http.StatusOK || !strings.Contains(body, `"status":"ready"`) || !strings.Contains(body, `"state":"DEGRADED"`) {
+		t.Fatalf("readiness response = %d %s", recorder.Code, body)
+	}
+}

tests/agent_execution/graph_runtime_test.go

@@ -0,0 +1,60 @@
+package agent_execution_test
+
+import (
+	"context"
+	"strconv"
+	"testing"
+
+	"git.cin.su/m/Assistant/contracts/work"
+	"git.cin.su/m/Assistant/modules/work/application"
+)
+
+type graphStore struct {
+	graphs  map[string]work.WorkGraph
+	queued  []string
+	intents []string
+}
+
+func (s *graphStore) SaveWorkGraph(_ context.Context, graph work.WorkGraph) error {
+	if s.graphs == nil {
+		s.graphs = map[string]work.WorkGraph{}
+	}
+	s.graphs[graph.ID+":"+itoa(graph.Revision)] = graph
+	return nil
+}
+func (s *graphStore) LoadWorkGraph(_ context.Context, id string, revision int64) (work.WorkGraph, error) {
+	return s.graphs[id+":"+itoa(revision)], nil
+}
+func (s *graphStore) SaveDispatchIntent(_ context.Context, intent work.AgentDispatchIntent) error {
+	s.intents = append(s.intents, intent.DispatchID)
+	return nil
+}
+func (s *graphStore) EnqueueDispatch(_ context.Context, item work.AgentDispatchQueueItem) error {
+	s.queued = append(s.queued, item.DispatchID)
+	return nil
+}
+
+func itoa(value int64) string {
+	return strconv.FormatInt(value, 10)
+}
+
+func TestGraphRuntimeSchedulesFanOutAndCompletesFanIn(t *testing.T) {
+	store := &graphStore{}
+	graph, ready, err := application.AdmitAndSchedule(context.Background(), store, application.GraphScheduleRequest{Graph: work.WorkGraph{ID: "g", Revision: 1, Budget: work.GraphBudget{MaxParallel: 2, MaxAttempts: 1}, Nodes: []work.GraphNode{{ID: "a", DispatchID: "d-a", IdempotencyKey: "a", State: work.GraphNodeReady}, {ID: "b", DispatchID: "d-b", IdempotencyKey: "b", State: work.GraphNodeReady}, {ID: "join", DispatchID: "d-join", IdempotencyKey: "join", State: work.GraphNodeBlocked}}, Edges: []work.GraphEdge{{From: "a", To: "join"}, {From: "b", To: "join"}}}, Intents: []work.AgentDispatchIntent{{DispatchID: "d-a"}, {DispatchID: "d-b"}, {DispatchID: "d-join"}}, Now: 1})
+	if err != nil || len(ready) != 2 || len(store.queued) != 2 {
+		t.Fatalf("graph=%+v ready=%v queued=%v err=%v", graph, ready, store.queued, err)
+	}
+	if _, err := application.CompleteGraphNode(context.Background(), store, "g", 2, "a", true); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := application.CompleteGraphNode(context.Background(), store, "g", 3, "b", true); err != nil {
+		t.Fatal(err)
+	}
+	if len(store.queued) != 3 || store.queued[2] != "d-join" {
+		t.Fatalf("fan-in dispatch was not enqueued: %v", store.queued)
+	}
+	completed, err := application.CompleteGraphNode(context.Background(), store, "g", 4, "join", true)
+	if err != nil || completed.State != work.GraphCompleted {
+		t.Fatalf("completed=%+v err=%v", completed, err)
+	}
+}

tests/research_contracts/truth_persistence_sql_test.go

@@ -0,0 +1,26 @@
+package research_contracts_test
+
+import (
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestTruthBundlePersistenceMigrationIsOwnedAndIdempotent(t *testing.T) {
+	payload, err := os.ReadFile("../../database/migrations/034_epistemics_truth_bundles.sql")
+	if err != nil {
+		t.Fatal(err)
+	}
+	script := string(payload)
+	for _, required := range []string{
+		"SET ROLE epistemics_role",
+		"CREATE TABLE IF NOT EXISTS epistemics.truth_bundles",
+		"UNIQUE (snapshot_id, excerpt_id, claim_id)",
+		"CREATE INDEX IF NOT EXISTS truth_bundles_source_time_idx",
+		"RESET ROLE",
+	} {
+		if !strings.Contains(script, required) {
+			t.Fatalf("migration missing %q", required)
+		}
+	}
+}

tests/research_runtime/http_connector_test.go

@@ -0,0 +1,20 @@
+package research_runtime_test
+
+import (
+	"context"
+	"testing"
+
+	researchadapter "git.cin.su/m/Assistant/adapters/research"
+	"git.cin.su/m/Assistant/contracts/epistemics"
+)
+
+func TestHTTPConnectorRejectsNonAllowlistedOrNonHTTPSSources(t *testing.T) {
+	connector := researchadapter.HTTPConnector{AllowedHosts: map[string]struct{}{"assistant.cin.su": {}}}
+	base := epistemics.WebResearchRequest{Plan: epistemics.ResearchPlan{ID: "plan", Question: "q", SourceClasses: []string{"web"}, FreshnessPolicy: "current"}, SourceID: "s", Statement: "statement"}
+	for _, url := range []string{"http://assistant.cin.su/healthz", "https://127.0.0.1/secret", "https://evil.example/"} {
+		base.URL = url
+		if _, err := connector.Collect(context.Background(), base, 10); err != researchadapter.ErrHTTPSourceRejected {
+			t.Fatalf("url=%s err=%v", url, err)
+		}
+	}
+}

tests/research_runtime/persisted_research_test.go

@@ -0,0 +1,29 @@
+package research_runtime_test
+
+import (
+	"context"
+	"testing"
+
+	"git.cin.su/m/Assistant/contracts/epistemics"
+	"git.cin.su/m/Assistant/modules/epistemics/application"
+)
+
+type store struct{ saved []epistemics.TruthBundle }
+
+func (s *store) SaveTruthBundle(_ context.Context, value epistemics.TruthBundle, _ int64) error {
+	s.saved = append(s.saved, value)
+	return nil
+}
+
+func TestEvaluateAndPersistStoresExactCandidateChain(t *testing.T) {
+	candidate := epistemics.ResearchCandidate{
+		PropositionKey: "p1", Statement: "fact", Source: epistemics.Source{ID: "source-1", Kind: "repo", Locator: "file:///repo/doc", Title: "doc"},
+		Snapshot: epistemics.SourceSnapshot{ID: "snapshot-1", SourceID: "source-1", ContentDigest: "sha256:x", CapturedAtUnixMillis: 10, Freshness: epistemics.FreshnessCurrent},
+		Excerpt:  epistemics.SourceExcerpt{ID: "excerpt-1", SnapshotID: "snapshot-1", Text: "fact", Start: 0, End: 4}, Evidence: epistemics.EvidenceObserved, Coverage: epistemics.CoverageComplete, Conflict: epistemics.ConflictNone,
+	}
+	store := &store{}
+	result, err := application.EvaluateAndPersist(context.Background(), epistemics.ResearchEvaluationRequest{Plan: epistemics.ResearchPlan{ID: "plan-1", Question: "q", SourceClasses: []string{"repo"}, FreshnessPolicy: "current"}, Candidates: []epistemics.ResearchCandidate{candidate}}, store, 20)
+	if err != nil || len(result.Findings) != 1 || len(store.saved) != 1 || store.saved[0].Excerpt.ID != candidate.Excerpt.ID || store.saved[0].Source.Locator != candidate.Source.Locator {
+		t.Fatalf("result=%#v saved=%#v err=%v", result, store.saved, err)
+	}
+}

tests/research_runtime/repository_connector_test.go

@@ -0,0 +1,28 @@
+package research_runtime_test
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	researchadapter "git.cin.su/m/Assistant/adapters/research"
+	"git.cin.su/m/Assistant/contracts/epistemics"
+)
+
+func TestRepositoryConnectorReturnsDigestAndExactExcerpt(t *testing.T) {
+	root := t.TempDir()
+	if err := os.WriteFile(filepath.Join(root, "evidence.txt"), []byte("exact repository evidence"), 0600); err != nil {
+		t.Fatal(err)
+	}
+	candidate, err := (researchadapter.RepositoryConnector{Root: root}).Collect(context.Background(), epistemics.RepositoryResearchRequest{Plan: epistemics.ResearchPlan{ID: "plan", Question: "q", SourceClasses: []string{"repository"}, FreshnessPolicy: "current"}, PropositionKey: "p", SourceID: "s", Path: "evidence.txt", Statement: "statement"}, 10)
+	if err != nil || candidate.Excerpt.Text != "exact repository evidence" || candidate.Snapshot.ContentDigest[:7] != "sha256:" || candidate.Excerpt.SnapshotID != candidate.Snapshot.ID {
+		t.Fatalf("candidate=%#v err=%v", candidate, err)
+	}
+}
+
+func TestRepositoryConnectorRejectsPathTraversal(t *testing.T) {
+	if _, err := (researchadapter.RepositoryConnector{Root: t.TempDir()}).Collect(context.Background(), epistemics.RepositoryResearchRequest{Plan: epistemics.ResearchPlan{ID: "plan", Question: "q", SourceClasses: []string{"repository"}, FreshnessPolicy: "current"}, SourceID: "s", Path: "../secret", Statement: "statement"}, 10); err != researchadapter.ErrRepositoryPathRejected {
+		t.Fatalf("path traversal error=%v", err)
+	}
+}