Contributions, contributor pages, fetch work and the CLI workerP3/r2 engine

README.md

@@ -238,6 +238,12 @@ cosift contents <url...> [-server URL] [-file PATH] [-text] [-json]
                                               hit /contents — single GET or batch POST
 cosift admin <stats|config|recrawl|recrawl-domain|reembed> [-server URL] [-token TOKEN] [-json]
                                               admin-protected operator endpoints
+cosift contribute -worker [-max-pages N] [-duration D] [-concurrency C] [-server URL]
+                                              fetch pages leased by the network and submit them
+                                              (bearer token required; see docs/COMMUNITY.md)
+cosift work purge --url URL | void --worker ID [-y] [-server URL] [-token TOKEN] [-json]
+                                              remove a forged page, or one worker's pending
+                                              attestations; a dry run without -y
 cosift stats [-backend sqlite|pebble]         doc / term counts + data dir (per-backend)
 cosift crawl-status [-hosts N] [-errors N] [-target N]
                                               live operator snapshot of an ongoing crawl: counts, frontier

cmd/cosift/cmd_work.go

@@ -0,0 +1,91 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/pilot-protocol/cosift/internal/config"
+)
+
+const workUsage = `usage: cosift work purge --url URL [-y] [-server URL] [-token TOKEN] [-json]
+       cosift work void --worker ID [-y] [-server URL] [-token TOKEN] [-json]
+
+purge removes what a forged URL indexed; void drops one worker's pending attestations.
+Without -y nothing changes and the command prints what it would do.`
+
+// runWork is `cosift work purge|void`: one call to a loopback /admin route of the running engine.
+func runWork(ctx context.Context, cfg *config.Config, args []string, stdout, stderr io.Writer) error {
+	if len(args) == 0 || (args[0] != "purge" && args[0] != "void") {
+		return &usageError{msg: workUsage}
+	}
+	op := args[0]
+	fs := flag.NewFlagSet("work "+op, flag.ContinueOnError)
+	fs.SetOutput(io.Discard)
+	server := fs.String("server", strings.Replace("http://"+cfg.Server.Addr, "0.0.0.0", "127.0.0.1", 1), "cosift server URL")
+	token := fs.String("token", "", "admin bearer token (defaults to COSIFT_ADMIN_TOKEN)")
+	jsonOut := fs.Bool("json", false, "print the raw JSON answer")
+	apply := fs.Bool("y", false, "apply the change")
+	target := fs.String("url", "", "the forged URL (purge)")
+	worker := fs.String("worker", "", "the worker id from the app's work_workers table (void)")
+	if err := fs.Parse(args[1:]); err != nil || fs.NArg() > 0 {
+		return &usageError{msg: workUsage}
+	}
+	path, body := "/admin/work-purge", map[string]any{"url": *target, "apply": *apply}
+	if op == "void" {
+		path, body = "/admin/work-void", map[string]any{"worker": *worker, "apply": *apply}
+	}
+	if (op == "purge" && *target == "") || (op == "void" && *worker == "") {
+		return &usageError{msg: workUsage}
+	}
+	if *token == "" {
+		*token = os.Getenv("COSIFT_ADMIN_TOKEN")
+	}
+	raw, _ := json.Marshal(body)
+	req, err := adminRequest(ctx, http.MethodPost, *server, path, *token, bytes.NewReader(raw))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	res, err := (&http.Client{Timeout: 5 * time.Minute}).Do(req)
+	if err != nil {
+		return fmt.Errorf("post %s: %w", *server, err)
+	}
+	defer res.Body.Close()
+	out, err := io.ReadAll(io.LimitReader(res.Body, 1<<20))
+	if err != nil {
+		return fmt.Errorf("read response: %w", err)
+	}
+	if res.StatusCode != http.StatusOK {
+		return fmt.Errorf("server %d: %s", res.StatusCode, bytes.TrimSpace(out))
+	}
+	if *jsonOut {
+		fmt.Fprintln(stdout, strings.TrimSpace(string(out)))
+		return nil
+	}
+	var m map[string]any
+	if err := json.Unmarshal(out, &m); err != nil {
+		return fmt.Errorf("decode: %w", err)
+	}
+	mode := "applied"
+	if !*apply {
+		mode = "dry run, nothing changed; add -y to apply"
+	}
+	fmt.Fprintf(stdout, "%s (%s)\n", op, mode)
+	for _, k := range workAnswerKeys[op] {
+		fmt.Fprintf(stdout, "  %-22s %v\n", k, m[k])
+	}
+	return nil
+}
+
+var workAnswerKeys = map[string][]string{
+	"purge": {"documents_deleted", "documents_kept", "records_deleted", "texts_deleted", "workers_settled", "parked_until"},
+	"void":  {"records_changed", "attestations_dropped", "records_deleted", "urls_parked"},
+}

cmd/cosift/cmd_work_test.go

@@ -0,0 +1,94 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/pilot-protocol/cosift/internal/config"
+)
+
+type workStub struct {
+	path, auth, ctype string
+	body              map[string]any
+	status            int
+	answer            string
+}
+
+func (s *workStub) serve(t *testing.T) string {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		s.path, s.auth, s.ctype = r.URL.Path, r.Header.Get("Authorization"), r.Header.Get("Content-Type")
+		b, _ := io.ReadAll(r.Body)
+		_ = json.Unmarshal(b, &s.body)
+		w.WriteHeader(s.status)
+		io.WriteString(w, s.answer)
+	}))
+	t.Cleanup(srv.Close)
+	return srv.URL
+}
+
+func runWorkCmd(url string, args ...string) (string, error) {
+	var out bytes.Buffer
+	err := runWork(context.Background(), &config.Config{}, append(args, "-server", url), &out, io.Discard)
+	return out.String(), err
+}
+
+func TestWorkPurgeCommandIsADryRunWithoutY(t *testing.T) {
+	t.Setenv("COSIFT_ADMIN_TOKEN", "env-token")
+	stub := &workStub{status: 200, answer: `{"documents_deleted":1,"documents_kept":0,"records_deleted":1,"texts_deleted":0,"workers_settled":2,"parked_until":"2026-10-27T10:00:00Z"}`}
+	url := stub.serve(t)
+	out, err := runWorkCmd(url, "purge", "-url", "https://a.example.org/x")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stub.path != "/admin/work-purge" || stub.auth != "Bearer env-token" || stub.ctype != "application/json" || stub.body["url"] != "https://a.example.org/x" || stub.body["apply"] != false || len(stub.body) != 2 {
+		t.Fatalf("%+v", stub)
+	}
+	if !strings.Contains(out, "dry run, nothing changed") || !strings.Contains(out, "documents_deleted") || !strings.Contains(out, "2026-10-27T10:00:00Z") {
+		t.Fatalf("%s", out)
+	}
+	out, err = runWorkCmd(url, "purge", "--url", "https://a.example.org/x", "-y", "-token", "flag-token")
+	if err != nil || stub.body["apply"] != true || stub.auth != "Bearer flag-token" || !strings.Contains(out, "purge (applied)") {
+		t.Fatalf("%+v %s %v", stub, out, err)
+	}
+}
+
+func TestWorkVoidCommand(t *testing.T) {
+	stub := &workStub{status: 200, answer: `{"records_changed":3,"attestations_dropped":3,"records_deleted":1,"urls_parked":2}`}
+	url := stub.serve(t)
+	worker := strings.Repeat("a", 32)
+	out, err := runWorkCmd(url, "void", "-worker", worker, "-y", "-json")
+	if err != nil || stub.path != "/admin/work-void" || stub.body["worker"] != worker || stub.body["apply"] != true {
+		t.Fatalf("%+v %v", stub, err)
+	}
+	if strings.TrimSpace(out) != stub.answer {
+		t.Fatalf("-json prints the raw answer: %s", out)
+	}
+	out, _ = runWorkCmd(url, "void", "-worker", worker)
+	if !strings.Contains(out, "urls_parked") || stub.body["apply"] != false {
+		t.Fatalf("%s", out)
+	}
+}
+
+func TestWorkCommandSurfacesRefusalsAndUsageErrors(t *testing.T) {
+	stub := &workStub{status: 401, answer: `{"error":"Unauthorized","status":401,"detail":"missing or invalid peer token"}`}
+	url := stub.serve(t)
+	if _, err := runWorkCmd(url, "purge", "-url", "https://a.example.org/x"); err == nil || !strings.Contains(err.Error(), "server 401") {
+		t.Fatalf("%v", err)
+	}
+	for _, args := range [][]string{{}, {"nuke"}, {"purge"}, {"void"}, {"purge", "-url", "u", "extra"}, {"void", "-bogus"}} {
+		var ue *usageError
+		if _, err := runWorkCmd(url, args...); !errors.As(err, &ue) {
+			t.Errorf("%v: %v", args, err)
+		}
+	}
+	if _, err := runWorkCmd("http://127.0.0.1:1", "purge", "-url", "u"); err == nil {
+		t.Fatal("an unreachable server is an error")
+	}
+}

cmd/cosift/community.go

@@ -23,6 +23,9 @@ import (
 	"github.com/pilot-protocol/cosift/internal/sharedaccount"
 )
 
+// openCommunity opens the community app; tests replace it.
+var openCommunity = community.Open
+
 func runCommunity(ctx context.Context, args []string) error {
 	fs := flag.NewFlagSet("community", flag.ContinueOnError)
 	addr := fs.String("addr", "127.0.0.1:7780", "listen address")
@@ -52,6 +55,14 @@ func runCommunity(ctx context.Context, args []string) error {
 	if err != nil {
 		return err
 	}
+	work, fetchCap, environment, err := communityWorkFromEnv()
+	if err != nil {
+		return err
+	}
+	moderationShare, err := communityModerationShareFromEnv()
+	if err != nil {
+		return err
+	}
 	var trusted []string
 	if *proxies != "" {
 		trusted = strings.Split(*proxies, ",")
@@ -69,7 +80,7 @@ func runCommunity(ctx context.Context, args []string) error {
 		defer client.Close()
 		provider = client
 	}
-	s, err := community.Open(community.Config{GAMeasurementID: os.Getenv("COSIFT_GA_MEASUREMENT_ID"), Shared: provider, SharedPasswordEnabled: os.Getenv("COSIFT_SHARED_PASSWORD_ENABLED") == "1", DataDir: *dir, Backend: *backend, PublicURL: *publicURL, BrokerKey: os.Getenv("COSIFT_COMMUNITY_BROKER_KEY"), EngineURL: os.Getenv("COSIFT_COMMUNITY_ENGINE_URL"), Awards: awards, TrustedProxies: trusted, GuestInterval: *guestInterval, MemberFreeRPM: *freeRPM, SearchRPM: *searchRPM, AnswerRPM: *answerRPM, ResearchPer10Min: *researchLimit, StripeSecretKey: os.Getenv("STRIPE_SECRET_KEY"), StripeWebhookSecret: os.Getenv("STRIPE_WEBHOOK_SECRET"), AllowTestPayments: os.Getenv("COSIFT_ALLOW_TEST_PAYMENTS") == "1", ContributionReward: rewardOn, ContributionRewardDailyCap: rewardCap, StripePortalConfigurationID: os.Getenv("COSIFT_STRIPE_PORTAL_CONFIGURATION_ID"), GSCVerification: os.Getenv("COSIFT_GSC_VERIFICATION"), Wiki: wikiConfig()})
+	s, err := openCommunity(community.Config{GAMeasurementID: os.Getenv("COSIFT_GA_MEASUREMENT_ID"), Shared: provider, SharedPasswordEnabled: os.Getenv("COSIFT_SHARED_PASSWORD_ENABLED") == "1", DataDir: *dir, Backend: *backend, PublicURL: *publicURL, BrokerKey: os.Getenv("COSIFT_COMMUNITY_BROKER_KEY"), EngineURL: os.Getenv("COSIFT_COMMUNITY_ENGINE_URL"), Awards: awards, Work: work, FetchDailyCap: fetchCap, ModerationsPerAccount: moderationShare, Environment: environment, TrustedProxies: trusted, GuestInterval: *guestInterval, MemberFreeRPM: *freeRPM, SearchRPM: *searchRPM, AnswerRPM: *answerRPM, ResearchPer10Min: *researchLimit, StripeSecretKey: os.Getenv("STRIPE_SECRET_KEY"), StripeWebhookSecret: os.Getenv("STRIPE_WEBHOOK_SECRET"), AllowTestPayments: os.Getenv("COSIFT_ALLOW_TEST_PAYMENTS") == "1", ContributionReward: rewardOn, ContributionRewardDailyCap: rewardCap, StripePortalConfigurationID: os.Getenv("COSIFT_STRIPE_PORTAL_CONFIGURATION_ID"), GSCVerification: os.Getenv("COSIFT_GSC_VERIFICATION"), Wiki: wikiConfig()})
 	if err != nil {
 		return err
 	}
@@ -109,6 +120,38 @@ func communityAwardsFromEnv() (bool, error) {
 	return false, fmt.Errorf("COSIFT_COMMUNITY_AWARDS must be 0 or 1")
 }
 
+// communityWorkFromEnv reads the work switch (0 by default), the per-account daily page cap (0 = default) and the environment label of the alerts.
+func communityWorkFromEnv() (enabled bool, fetchCap int, environment string, err error) {
+	switch strings.TrimSpace(os.Getenv("COSIFT_COMMUNITY_WORK")) {
+	case "", "0":
+	case "1":
+		enabled = true
+	default:
+		return false, 0, "", fmt.Errorf("COSIFT_COMMUNITY_WORK must be 0 or 1")
+	}
+	if v := strings.TrimSpace(os.Getenv("COSIFT_COMMUNITY_FETCH_DAILY_CAP")); v != "" {
+		n, convErr := strconv.Atoi(v)
+		if convErr != nil || n < 1 || n > 10000 {
+			return false, 0, "", fmt.Errorf("COSIFT_COMMUNITY_FETCH_DAILY_CAP must be an integer from 1 to 10000")
+		}
+		fetchCap = n
+	}
+	return enabled, fetchCap, strings.TrimSpace(os.Getenv("COSIFT_ENV")), nil
+}
+
+// communityModerationShareFromEnv reads the content checks one account may use per UTC day (0 = default).
+func communityModerationShareFromEnv() (int, error) {
+	v := strings.TrimSpace(os.Getenv("COSIFT_COMMUNITY_MODERATIONS_PER_ACCOUNT"))
+	if v == "" {
+		return 0, nil
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil || n < 1 || n > 10000 {
+		return 0, fmt.Errorf("COSIFT_COMMUNITY_MODERATIONS_PER_ACCOUNT must be an integer from 1 to 10000")
+	}
+	return n, nil
+}
+
 // communityRewardFromEnv reads the contribution-reward switch (off by default) and the per-account daily cap (0 = default).
 func communityRewardFromEnv() (enabled bool, dailyCap int, err error) {
 	switch strings.ToLower(strings.TrimSpace(os.Getenv("COSIFT_COMMUNITY_CONTRIBUTION_REWARD"))) {
@@ -146,14 +189,25 @@ func runContributeConfigured(ctx context.Context, cfg *config.Config, args []str
 	login := fs.Bool("login", false, "save an authenticated CLI session")
 	logout := fs.Bool("logout", false, "revoke and delete the saved CLI session")
 	guest := fs.Bool("guest", false, "perform Search, Answer or Research without login; contributions require login")
+	workerMode := fs.Bool("worker", false, "fetch blocked pages leased by the network and submit them (needs your bearer token)")
+	maxPages := fs.Int("max-pages", 100, "worker: stop after claiming this many pages (0 = no page limit)")
+	duration := fs.Duration("duration", 30*time.Minute, "worker: stop claiming after this long (0 = no time limit)")
+	concurrency := fs.Int("concurrency", 4, "worker: leases in flight at once, 1 to 32")
 	if err := fs.Parse(args); err != nil {
 		return err
 	}
-	var explicitServer, explicitSession bool
+	var explicitServer, explicitSession, workerFlag bool
 	fs.Visit(func(f *flag.Flag) {
 		explicitServer = explicitServer || f.Name == "server"
 		explicitSession = explicitSession || f.Name == "session-file"
+		workerFlag = workerFlag || f.Name == "max-pages" || f.Name == "duration" || f.Name == "concurrency"
 	})
+	if workerFlag && !*workerMode {
+		return fmt.Errorf("-max-pages, -duration and -concurrency require -worker")
+	}
+	if *workerMode && (*login || *logout || *guest || *requestMode || *local || *credits || *file != "" || *query != "" || fs.NArg() > 0) {
+		return fmt.Errorf("-worker cannot be combined with other operations")
+	}
 	// The installer can provision this one well-known session. Explicit
 	// credentials and guest mode always win; no harness config is inspected.
 	if !*guest && !*login && !explicitSession && *sessionFile == "" && os.Getenv("COSIFT_TOKEN") == "" && *email == "" && os.Getenv("COSIFT_PASSWORD") == "" {
@@ -188,6 +242,13 @@ func runContributeConfigured(ctx context.Context, cfg *config.Config, args []str
 	if token != "" && *sessionFile != "" && !*login {
 		return fmt.Errorf("choose COSIFT_TOKEN or a saved session, not both")
 	}
+	if *workerMode {
+		bearer, err := workerBearer(token, *sessionFile, strings.TrimRight(*server, "/"), password != "" || *email != "")
+		if err != nil {
+			return err
+		}
+		return runWorkerMode(ctx, strings.TrimRight(*server, "/"), bearer, workerOptions{MaxPages: *maxPages, Duration: *duration, Concurrency: *concurrency}, os.Stdout, os.Stderr)
+	}
 	if token == "" && !*guest && (*login || *sessionFile == "") && ((*email == "") != (password == "")) {
 		return fmt.Errorf("set both COSIFT_EMAIL and COSIFT_PASSWORD, or use -guest for retrieval")
 	}

cmd/cosift/community_local.go

@@ -16,7 +16,7 @@ import (
 )
 
 func indexLocalContributions(ctx context.Context, cfg *config.Config, urls []string) ([]*crawler.LocalArtifact, error) {
-	return indexLocalContributionsWithClient(ctx, cfg, urls, crawler.PublicHTTPClient(30*time.Second))
+	return indexLocalContributionsWithClient(ctx, cfg, urls, crawler.BrowserClient(30*time.Second))
 }
 
 func indexLocalContributionsWithClient(ctx context.Context, cfg *config.Config, urls []string, client *http.Client) ([]*crawler.LocalArtifact, error) {
@@ -51,7 +51,7 @@ func indexLocalContributionsWithClient(ctx context.Context, cfg *config.Config,
 			case <-timer.C:
 			}
 		}
-		page, err := crawler.FetchOne(ctx, client, "Cosift-Community/1.0", raw, 2<<20)
+		page, err := fetchAndExtract(ctx, client, crawler.BrowserHeaders(), raw, 2<<20)
 		if err != nil {
 			return nil, err
 		}

cmd/cosift/community_local_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 
 	"github.com/pilot-protocol/cosift/internal/config"
+	"github.com/pilot-protocol/cosift/internal/crawler"
 	"github.com/pilot-protocol/cosift/internal/store"
 )
 
@@ -55,3 +56,41 @@ func TestLocalIndexPersistsTextAndEmbeddings(t *testing.T) {
 		t.Fatal("local document was not indexed")
 	}
 }
+
+func TestLocalIndexFetchesThroughTheSharedBrowserFetch(t *testing.T) {
+	emb := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var body struct {
+			Input []string `json:"input"`
+		}
+		json.NewDecoder(r.Body).Decode(&body)
+		data := []any{}
+		for i := range body.Input {
+			data = append(data, map[string]any{"index": i, "embedding": []float32{1, 2}})
+		}
+		json.NewEncoder(w).Encode(map[string]any{"data": data})
+	}))
+	defer emb.Close()
+	cfg := &config.Config{DataDir: t.TempDir(), Embeddings: config.Embeddings{URL: emb.URL, Model: "test", Dim: 2}}
+	var pageUA string
+	client := &http.Client{Transport: localPageTransport(func(r *http.Request) (*http.Response, error) {
+		status, body := 200, `<html><title>Guide</title><article>`+strings.Repeat("This is useful public scientific documentation. ", 5)+`</article></html>`
+		switch r.URL.Path {
+		case "/robots.txt":
+			body = "User-agent: *\nAllow: /"
+		case "/gone":
+			status = 404
+		default:
+			pageUA = r.Header.Get("User-Agent")
+		}
+		return &http.Response{StatusCode: status, Header: http.Header{"Content-Type": []string{"text/html"}}, Body: io.NopCloser(strings.NewReader(body)), Request: r}, nil
+	})}
+	if _, err := indexLocalContributionsWithClient(context.Background(), cfg, []string{"https://example.com/guide"}, client); err != nil {
+		t.Fatal(err)
+	}
+	if pageUA != crawler.BrowserUserAgent {
+		t.Fatalf("the page was fetched as %q", pageUA)
+	}
+	if _, err := indexLocalContributionsWithClient(context.Background(), cfg, []string{"https://example.com/gone"}, client); err == nil || !strings.Contains(err.Error(), "404") {
+		t.Fatalf("a 404 page was indexed: %v", err)
+	}
+}