Tasks

Integrate an agent

Publish a taskable agent, process assigned tasks, and report status using the Lattice SDK

A taskable agent is an asset, or a group of assets, that listens for tasks assigned to it in Lattice, executes them, and reports its progress back. This guide walks through the full agent workflow: publishing a taskable agent, streaming and parsing assigned tasks, updating task status, and handling cancellation requests.

An agent calls the following API to listen for tasks assigned to it in Lattice:

  • StreamAsAgent — For monitoring tasks routed to the agent using REST.
  • ListenAsAgent — For monitoring tasks routed to the agent using gRPC.

Before you begin

  • To publish taskable entities, and subscribe to tasks, set up your Lattice environment.
  • Familiarize yourself with entities and different entity types.
  • Review how to define a task. This guide uses the Objective task definition from that page as its example.
gRPC authentication

If you are using gRPC with client credentials, set up the token refresh module before running the examples on this page.

Publish a taskable agent

An asset is an entity under your control, or under the control of another operator or system. Assets may accept tasks such as search or tracking. An agent is an asset, or a group of assets, that can complete a specific set of defined tasks.

To publish an agent, do the following:

1

Define a TaskCatalog

The entity model’s TaskCatalog component defines the tasks that an asset can execute.

For example, if your asset supports the Objective task, publish an asset entity that includes it in its catalog, and listen for and execute that task. The taskSpecificationUrl is the type URL of the task definition you authored and published to the Schema Registry:

taskCatalog
1  "taskCatalog": {
2      // Define the tasks the asset can perform.
3      "taskDefinitions": [
4          {
5              // Set the task specification URL to your Objective definition.
6              "taskSpecificationUrl": "type.googleapis.com/org.example.reconnaissance.v1beta.Objective",
7          }
8      ]
9  }
2

Publish the agent

Use the PublishEntity method to publish a taskable agent:

1package main
2
3import (
4	"context"
5	"fmt"
6	"net/http"
7	"os"
8	"time"
9
10	Lattice "github.com/anduril/lattice-sdk-go/v4"
11	"github.com/anduril/lattice-sdk-go/v4/client"
12	"github.com/anduril/lattice-sdk-go/v4/option"
13	"github.com/google/uuid"
14)
15
16func main() {
17	// Get environment variables
18	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
19	clientSecret := os.Getenv("LATTICE_CLIENT_SECRET")
20	clientId := os.Getenv("LATTICE_CLIENT_ID")
21
22	// Remove sandboxesToken from the following statements if you are not developing on Sandboxes
23	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
24
25	// Check required environment variables
26	if latticeEndpoint == "" || clientId == "" || clientSecret == "" || sandboxesToken == "" {
27		fmt.Println("Missing required environment variables")
28		os.Exit(1)
29	}
30
31	// Initialize headers for sandbox authorization
32	headers := http.Header{}
33	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
34
35	// Create the client
36	LatticeClient := client.NewClient(
37		option.WithClientCredentials(clientId, clientSecret),
38		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
39		option.WithHTTPHeader(headers),
40	)
41
42	// Generate a unique ID for the entity
43	entityId := uuid.New().String()
44
45	// Get creation time
46	creationTime := time.Now().UTC()
47
48	// Continuously publish the entity
49	for {
50		latestTimestamp := time.Now().UTC()
51		ctx := context.Background()
52
53		// Create entity to publish
54		entity := Lattice.Entity{
55			EntityID:    &entityId,
56			Description: Lattice.String("Friendly drone asset"),
57			Aliases: &Lattice.Aliases{
58				Name: Lattice.String("Drone 1"),
59			},
60			IsLive:      Lattice.Bool(true),
61			CreatedTime: Lattice.Time(creationTime),
62			ExpiryTime:  Lattice.Time(latestTimestamp.Add(1 * time.Minute)),
63			Ontology: &Lattice.Ontology{
64				Template:     Lattice.OntologyTemplateTemplateAsset.Ptr(),
65				PlatformType: Lattice.String("UAV"),
66			},
67			MilView: &Lattice.MilView{
68				Disposition: Lattice.MilViewDispositionDispositionFriendly.Ptr(),
69				Environment: Lattice.MilViewEnvironmentEnvironmentAir.Ptr(),
70			},
71			Location: &Lattice.Location{
72				Position: &Lattice.Position{
73					LatitudeDegrees:   Lattice.Float64(50.91402185768586),
74					LongitudeDegrees:  Lattice.Float64(0.79203612077257),
75					AltitudeAsfMeters: Lattice.Float64(1000),
76				},
77			},
78			Provenance: &Lattice.Provenance{
79				IntegrationName:  Lattice.String("your_integration_name"),
80				DataType:         Lattice.String("your_data_type"),
81				SourceUpdateTime: Lattice.Time(latestTimestamp),
82			},
83			Health: &Lattice.Health{
84				ConnectionStatus: Lattice.HealthConnectionStatusConnectionStatusOnline.Ptr(),
85				HealthStatus:     Lattice.HealthHealthStatusHealthStatusHealthy.Ptr(),
86				UpdateTime:       Lattice.Time(latestTimestamp),
87			},
88			TaskCatalog: &Lattice.TaskCatalog{
89				TaskDefinitions: []*Lattice.TaskDefinition{
90					{TaskSpecificationURL: Lattice.String("type.googleapis.com/anduril.tasks.v2.VisualId")},
91					{TaskSpecificationURL: Lattice.String("type.googleapis.com/anduril.tasks.v2.Investigate")},
92				},
93			},
94		}
95
96		// Publish the entity
97		_, err := LatticeClient.Entities.PublishEntity(ctx, &entity)
98
99		// Handle errors
100		if err != nil {
101			fmt.Printf("Error publishing entity: %v\n", err)
102		} else {
103			fmt.Println("Published asset: " + entityId)
104		}
105
106		// Wait before next request
107		time.Sleep(5 * time.Second)
108	}
109}

If successful, you see the entity ID in the console:

$2025-07-16T02:50:04.694Z [INFO]: Published asset: <entity-id>

Process assigned tasks

Once an agent is published, the integration should open a stream to receive tasks routed to it. Each task carries a specification — a google.protobuf.Any that holds the task’s data. To act on a task, the agent reads the metadata off the request, then decodes the specification into the task type it advertised in its catalog.

1

Stream tasks as an agent

Use StreamAsAgent, or ListenAsAgent for gRPC, to open the stream. When an executeRequest arrives, the agent reads the task ID and status version, parses the task data, then reports that it has started the task.

Task delivery over a stream does not wait for the agent to acknowledge receipt, so a silently dropped connection might result in your agent silently skipping a task. To get a positive signal that the connection is alive, set heartbeatIntervalMs when you open the stream:

heartbeatIntervalMs
Number

The interval in milliseconds at which Lattice sends heartbeat events on the stream. For StreamAsAgent, the default is 30000 ms, and the minimum is 1000 ms. Smaller values are raised to 1000 ms. For ListenAsAgent, heartbeats are disabled unless you set this field.

When set, Lattice sends a heartbeat at the specified interval, and your agent can treat a missing heartbeat as a dropped connection and reconnect. Heartbeats arrive on the same stream as task requests and carry only a timestamp, so the receive loop skips them rather than treating them as tasks.

Replace AGENT_ID with the ID of the agent you want to task. If you are developing on Sandboxes, replace this with the following simulated asset: Demo-Sim-Asset1:

1package main
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"log"
9	"net/http"
10	"os"
11	"time"
12
13	Lattice "github.com/anduril/lattice-sdk-go/v4"
14	"github.com/anduril/lattice-sdk-go/v4/client"
15	"github.com/anduril/lattice-sdk-go/v4/option"
16)
17
18// Type URL of the Objective task defined in objective.proto.
19const objectiveType = "type.googleapis.com/org.example.reconnaissance.v1beta.Objective"
20
21// parseObjective parses the Objective carried in the task's specification. Over
22// REST, the specification is a google.protobuf.Any represented as JSON: a Type
23// URL alongside the task's flattened fields, which the SDK surfaces as
24// ExtraProperties.
25func parseObjective(specification *Lattice.GoogleProtobufAny) {
26	if specification == nil || specification.Type == nil || *specification.Type != objectiveType {
27		log.Printf("Unsupported task type: %v", specification.GetType())
28		return
29	}
30
31	// Objective is a oneof: read whichever target the operator set.
32	fields := specification.ExtraProperties
33	if entityID, ok := fields["entityId"].(string); ok && entityID != "" {
34		log.Printf("  Objective: entity %s", entityID)
35	} else if lla, ok := fields["lla"].(map[string]interface{}); ok {
36		log.Printf("  Objective: LLA (lat %v, lon %v, alt %vm)",
37			lla["latitudeDegrees"], lla["longitudeDegrees"], lla["altitudeHaeM"])
38	} else {
39		log.Printf("  Objective: no target set")
40	}
41}
42
43func main() {
44	// Get environment variables
45	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
46	clientSecret := os.Getenv("LATTICE_CLIENT_SECRET")
47	clientId := os.Getenv("LATTICE_CLIENT_ID")
48	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
49
50	// Check required environment variables
51	if latticeEndpoint == "" || clientId == "" || clientSecret == "" || sandboxesToken == "" {
52		fmt.Println("Missing required environment variables")
53		os.Exit(1)
54	}
55
56	// Initialize headers for sandbox authorization
57	headers := http.Header{}
58	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
59	// Create the client
60	LatticeClient := client.NewClient(
61		option.WithClientCredentials(clientId, clientSecret),
62		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
63		option.WithHTTPHeader(headers),
64	)
65
66	// Set the entity ID to listen for tasks
67	entityId := "<AGENT_ID>"
68	fmt.Printf("Streaming tasks for entity %s...\n", entityId)
69
70	// Create context for the request
71	ctx := context.Background()
72
73	// Create agent stream request. Set HeartbeatIntervalMs so Lattice sends
74	// periodic heartbeats; a missing heartbeat signals a dropped connection.
75	heartbeatIntervalMs := 30000
76	agentStreamRequest := Lattice.AgentStreamRequest{
77		AgentSelector: &Lattice.EntityIDsSelector{
78			EntityIDs: []string{entityId},
79		},
80		HeartbeatIntervalMs: &heartbeatIntervalMs,
81	}
82
83	// Stream tasks
84	stream, err := LatticeClient.Tasks.StreamAsAgent(ctx, &agentStreamRequest)
85	if err != nil {
86		fmt.Printf("Error streaming tasks: %v\n", err)
87		os.Exit(1)
88	}
89
90	// Process stream events
91	for {
92		select {
93		case <-ctx.Done():
94			log.Printf("Context canceled: %v", ctx.Err())
95			return
96		default:
97			// Continue processing
98		}
99
100		event, err := stream.Recv()
101
102		if errors.Is(err, io.EOF) {
103			log.Println("Stream completed successfully.")
104			return
105		}
106
107		if err != nil {
108			log.Printf("Error receiving message: %v", err)
109			continue
110		}
111
112		if event.Event == "heartbeat" {
113			timestamp := *event.Heartbeat.Timestamp
114			log.Printf("Heartbeat: %s", timestamp)
115		} else {
116			request := event.GetAgentRequest()
117			if executeRequest := request.GetExecuteRequest(); executeRequest != nil {
118				task := executeRequest.GetTask()
119				if task != nil {
120					taskId := *task.GetVersion().GetTaskID()
121					taskStatusVersion := *task.GetVersion().GetStatusVersion()
122					description := *task.GetDescription()
123
124					log.Printf("Starting task %s, version %d: %s", taskId, taskStatusVersion, description)
125
126					// Parse the Objective the operator sent with the task.
127					parseObjective(task.GetSpecification())
128
129					// Update task status to STATUS_EXECUTING
130					result, err := startTask(ctx, LatticeClient, taskId, int(taskStatusVersion), entityId)
131					if err != nil {
132						log.Printf("Error starting task: %v", err)
133						continue
134					}
135
136					log.Printf("Started task with status version: %d", *result.StatusVersion)
137				}
138			} else if completeRequest := request.GetCompleteRequest(); completeRequest != nil {
139				taskToComplete := completeRequest.GetTaskID()
140				if taskToComplete != nil {
141					log.Printf("Completing task: %s", *taskToComplete)
142				}
143			} else if cancelRequest := request.GetCancelRequest(); cancelRequest != nil {
144				taskToCancel := cancelRequest.GetTaskID()
145				if taskToCancel != nil {
146					log.Printf("Cancelling task: %s", *taskToCancel)
147				}
148			}
149		}
150
151		// Sleep briefly to prevent tight looping
152		time.Sleep(100 * time.Millisecond)
153	}
154}
155
156// startTask updates the task status to STATUS_EXECUTING
157func startTask(ctx context.Context, client *client.Client, taskId string, taskStatusVersion int, agentEntityId string) (*Lattice.TaskVersion, error) {
158	// Increment status version for the update
159	taskStatusVersion++
160
161	// Create system principal with the agent entity ID
162	principal := Lattice.Principal{
163		System: &Lattice.System{
164			EntityID: &agentEntityId,
165		},
166	}
167
168	taskStatus := Lattice.TaskStatus{
169		Status: Lattice.TaskStatusStatusStatusExecuting.Ptr(),
170	}
171
172	// Create task status update request
173	taskStatusUpdate := Lattice.TaskStatusUpdate{
174		TaskID:        taskId,
175		StatusVersion: &taskStatusVersion,
176		NewStatus:     &taskStatus,
177		Author:        &principal,
178	}
179
180	// Call the UpdateTaskStatus API
181	task, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
182	if err != nil {
183		return nil, fmt.Errorf("error updating task status: %w", err)
184	}
185
186	return task.Version, nil
187}

If successful, you see the following output:

$Listening for tasks for entity with ID <entity-id>...

An agent should also handle stream interruptions gracefully. If the connection to Lattice fails, the stream closes. Handle stream errors and reconnect so that your agent keeps its subscription to task updates through transient network issues.

For more information, see Retry connections.

This example reports STATUS_EXECUTING as soon as a task arrives, but an agent isn’t obligated to accept every task.

Before executing the task, validate that the agent can actually perform the task — for example, that the requested parameters are within its capabilities. If it can’t, the agent can reject the task by reporting STATUS_DONE_NOT_OK with a TaskError of ERROR_CODE_REJECTED instead of starting execution.

2

Parse the task specification

Decode the task’s specification to read the Objective the operator sent. Over REST, the specification is delivered as JSON — a type URL alongside the task’s flattened fields. Over gRPC, it’s a protobuf Any that you unpack into the Objective message generated from your message. Check the type URL before reading the fields, since an agent’s catalog can advertise more than one task type. Because Objective is a oneof, inspect which target is set — an entity_id or an lla point:

1package main
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"log"
9	"net/http"
10	"os"
11	"time"
12
13	Lattice "github.com/anduril/lattice-sdk-go/v4"
14	"github.com/anduril/lattice-sdk-go/v4/client"
15	"github.com/anduril/lattice-sdk-go/v4/option"
16)
17
18// Type URL of the Objective task defined in objective.proto.
19const objectiveType = "type.googleapis.com/org.example.reconnaissance.v1beta.Objective"
20
21// parseObjective parses the Objective carried in the task's specification. Over
22// REST, the specification is a google.protobuf.Any represented as JSON: a Type
23// URL alongside the task's flattened fields, which the SDK surfaces as
24// ExtraProperties.
25func parseObjective(specification *Lattice.GoogleProtobufAny) {
26	if specification == nil || specification.Type == nil || *specification.Type != objectiveType {
27		log.Printf("Unsupported task type: %v", specification.GetType())
28		return
29	}
30
31	// Objective is a oneof: read whichever target the operator set.
32	fields := specification.ExtraProperties
33	if entityID, ok := fields["entityId"].(string); ok && entityID != "" {
34		log.Printf("  Objective: entity %s", entityID)
35	} else if lla, ok := fields["lla"].(map[string]interface{}); ok {
36		log.Printf("  Objective: LLA (lat %v, lon %v, alt %vm)",
37			lla["latitudeDegrees"], lla["longitudeDegrees"], lla["altitudeHaeM"])
38	} else {
39		log.Printf("  Objective: no target set")
40	}
41}
42
43func main() {
44	// Get environment variables
45	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
46	clientSecret := os.Getenv("LATTICE_CLIENT_SECRET")
47	clientId := os.Getenv("LATTICE_CLIENT_ID")
48	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
49
50	// Check required environment variables
51	if latticeEndpoint == "" || clientId == "" || clientSecret == "" || sandboxesToken == "" {
52		fmt.Println("Missing required environment variables")
53		os.Exit(1)
54	}
55
56	// Initialize headers for sandbox authorization
57	headers := http.Header{}
58	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
59	// Create the client
60	LatticeClient := client.NewClient(
61		option.WithClientCredentials(clientId, clientSecret),
62		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
63		option.WithHTTPHeader(headers),
64	)
65
66	// Set the entity ID to listen for tasks
67	entityId := "<AGENT_ID>"
68	fmt.Printf("Streaming tasks for entity %s...\n", entityId)
69
70	// Create context for the request
71	ctx := context.Background()
72
73	// Create agent stream request. Set HeartbeatIntervalMs so Lattice sends
74	// periodic heartbeats; a missing heartbeat signals a dropped connection.
75	heartbeatIntervalMs := 30000
76	agentStreamRequest := Lattice.AgentStreamRequest{
77		AgentSelector: &Lattice.EntityIDsSelector{
78			EntityIDs: []string{entityId},
79		},
80		HeartbeatIntervalMs: &heartbeatIntervalMs,
81	}
82
83	// Stream tasks
84	stream, err := LatticeClient.Tasks.StreamAsAgent(ctx, &agentStreamRequest)
85	if err != nil {
86		fmt.Printf("Error streaming tasks: %v\n", err)
87		os.Exit(1)
88	}
89
90	// Process stream events
91	for {
92		select {
93		case <-ctx.Done():
94			log.Printf("Context canceled: %v", ctx.Err())
95			return
96		default:
97			// Continue processing
98		}
99
100		event, err := stream.Recv()
101
102		if errors.Is(err, io.EOF) {
103			log.Println("Stream completed successfully.")
104			return
105		}
106
107		if err != nil {
108			log.Printf("Error receiving message: %v", err)
109			continue
110		}
111
112		if event.Event == "heartbeat" {
113			timestamp := *event.Heartbeat.Timestamp
114			log.Printf("Heartbeat: %s", timestamp)
115		} else {
116			request := event.GetAgentRequest()
117			if executeRequest := request.GetExecuteRequest(); executeRequest != nil {
118				task := executeRequest.GetTask()
119				if task != nil {
120					taskId := *task.GetVersion().GetTaskID()
121					taskStatusVersion := *task.GetVersion().GetStatusVersion()
122					description := *task.GetDescription()
123
124					log.Printf("Starting task %s, version %d: %s", taskId, taskStatusVersion, description)
125
126					// Parse the Objective the operator sent with the task.
127					parseObjective(task.GetSpecification())
128
129					// Update task status to STATUS_EXECUTING
130					result, err := startTask(ctx, LatticeClient, taskId, int(taskStatusVersion), entityId)
131					if err != nil {
132						log.Printf("Error starting task: %v", err)
133						continue
134					}
135
136					log.Printf("Started task with status version: %d", *result.StatusVersion)
137				}
138			} else if completeRequest := request.GetCompleteRequest(); completeRequest != nil {
139				taskToComplete := completeRequest.GetTaskID()
140				if taskToComplete != nil {
141					log.Printf("Completing task: %s", *taskToComplete)
142				}
143			} else if cancelRequest := request.GetCancelRequest(); cancelRequest != nil {
144				taskToCancel := cancelRequest.GetTaskID()
145				if taskToCancel != nil {
146					log.Printf("Cancelling task: %s", *taskToCancel)
147				}
148			}
149		}
150
151		// Sleep briefly to prevent tight looping
152		time.Sleep(100 * time.Millisecond)
153	}
154}
155
156// startTask updates the task status to STATUS_EXECUTING
157func startTask(ctx context.Context, client *client.Client, taskId string, taskStatusVersion int, agentEntityId string) (*Lattice.TaskVersion, error) {
158	// Increment status version for the update
159	taskStatusVersion++
160
161	// Create system principal with the agent entity ID
162	principal := Lattice.Principal{
163		System: &Lattice.System{
164			EntityID: &agentEntityId,
165		},
166	}
167
168	taskStatus := Lattice.TaskStatus{
169		Status: Lattice.TaskStatusStatusStatusExecuting.Ptr(),
170	}
171
172	// Create task status update request
173	taskStatusUpdate := Lattice.TaskStatusUpdate{
174		TaskID:        taskId,
175		StatusVersion: &taskStatusVersion,
176		NewStatus:     &taskStatus,
177		Author:        &principal,
178	}
179
180	// Call the UpdateTaskStatus API
181	task, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
182	if err != nil {
183		return nil, fmt.Errorf("error updating task status: %w", err)
184	}
185
186	return task.Version, nil
187}

The gRPC examples import an Objective type generated from your own objective.proto. Publish your schema to the Schema Registry, then generate the language bindings before running the agent.

Task and verify the task handler

The following steps aren’t required to implement tasking — they show how to exercise your task handler end to end in a development environment. Assign a task to your agent, then confirm it receives, parses, and updates the task.

1

Assign a task using the UI

In most cases an operator uses the Lattice UI to task your agent. To test the agent’s task handler in Lattice Sandboxes, do the following:

  1. Open your environment’s Lattice UI, and choose an asset from the Assets panel. On Sandboxes, choose Demo-Sim-Asset1.
  2. From the entity pane, choose Task, then select the Objective task.
  3. From the Task Details panel on the right hand side, set the objective — either an entity to reconnoiter or a lat/lon/altitude point — then choose Execute Task. This is the same field your agent reads from the task specification.
2

Verify the task status

If successful, you see the parsed task data and the updated status version in your local development console:

$Starting task <task-id>, version 1: Reconnaissance objective
$  Objective: LLA (lat 37.7749, lon -122.4194, alt 100.0m)
$Started task with status version: 2.

The agent has successfully parsed the task data, updated the task status, and incremented the task status version.

Update the status of a task

As an agent makes progress, it reports real-time updates to Lattice with UpdateTaskStatus, incrementing the task’s statusVersion on each update. In Lattice, tasks move through the following states:

1

STATUS_SENT

Lattice automatically starts with STATUS_SENT when it sends a task to an agent:

1"task": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 1
6    },
7    "status": {
8        "status": "STATUS_SENT"
9    }
10}
2

STATUS_MACHINE_RECEIPT

The agent then responds back with status STATUS_MACHINE_RECEIPT, indicating that the task has been received, and incrementing statusVersion accordingly:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 2
6    },
7    "status": {
8        "status": "STATUS_MACHINE_RECEIPT"
9    }
10}
3

STATUS_ACK

When the agent is ready to acknowledge the task, it does so using STATUS_ACK, and again increments statusVersion:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 3
6    },
7    "status": {
8        "status": "STATUS_ACK"
9    }
10}
4

STATUS_WILCO

The agent confirms it intends to execute the task using STATUS_WILCO:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 4
6    },
7    "status": {
8        "status": "STATUS_WILCO"
9    }
10}
5

STATUS_EXECUTING

As the agent begins to actively execute the task, it indicates this by reporting STATUS_EXECUTING back to Lattice:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 5
6    },
7    "status": {
8        "status": "STATUS_EXECUTING"
9    }
10}
6

STATUS_DONE_OK

Finally, when the agent reaches a terminal state and completes the task successfully, it reports STATUS_DONE_OK. The agent can reach this state on its own, such as when its logic determines that the task’s objective is met, or in response to an operator-initiated request for task completion:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 6
6    },
7    "status": {
8        "status": "STATUS_DONE_OK"
9    }
10}
7

STATUS_DONE_NOT_OK

If the agent reaches a terminal state but does not complete the task successfully, it reports STATUS_DONE_NOT_OK. The agent can reach this state on its own, such as when its logic determines that the task can’t be completed, or in response to an operator-initiated request for task completion or cancellation. The agent should include a descriptive TaskError when reporting STATUS_DONE_NOT_OK:

1"statusUpdate": {
2    "version": {
3        "taskId": "my-task",
4        "definitionVersion": 1,
5        "statusVersion": 7
6    },
7    "status": {
8        "status": "STATUS_DONE_NOT_OK",
9        "taskError": {
10        "code": "ERROR_CODE_FAILED",
11        "message": "The asset failed task execution due to an internal error.",
12        }
13    }
14}

In this example, the message indicates that the agent encountered an internal error during task execution. You can add more descriptive errors to help the operator troubleshoot the issue accordingly.

The STATUS_DONE_OK and STATUS_DONE_NOT_OK statuses are considered terminal states. Once a task reaches either state, it’s complete and cannot be updated.

Handle task cancellation

An operator can request that a task be cancelled while the agent is executing it. When a task has already been sent to an agent, Lattice routes the CancelTask request to that agent, which decides whether to accept or reject it. For the operator’s side of this workflow, see Cancel tasks.

1

Set up a task processor

Create a task processor that listens for tasks assigned to your agent and handles cancellation requests. For demonstration purposes, this example uses a TASK_ACTIVE environment variable to control whether the agent accepts or rejects cancellations:

1package main
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"log"
9	"net/http"
10	"os"
11	"strings"
12	"time"
13
14	Lattice "github.com/anduril/lattice-sdk-go/v4"
15	"github.com/anduril/lattice-sdk-go/v4/client"
16	"github.com/anduril/lattice-sdk-go/v4/option"
17)
18
19// Sets whether the task is currently being processed. If so, it cannot be cancelled.
20var taskActive bool
21
22func main() {
23	// Get environment variables
24	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
25	clientSecret := os.Getenv("LATTICE_CLIENT_SECRET")
26	clientId := os.Getenv("LATTICE_CLIENT_ID")
27	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
28	taskActiveStr := os.Getenv("TASK_ACTIVE")
29
30	// Check required environment variables
31	if latticeEndpoint == "" || clientId == "" || clientSecret == "" || sandboxesToken == "" {
32		fmt.Println("Missing required environment variables")
33		os.Exit(1)
34	}
35
36	// Parse TASK_ACTIVE environment variable
37	taskActive = strings.ToLower(taskActiveStr) == "true"
38
39	// Initialize headers for sandbox authorization
40	headers := http.Header{}
41	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
42
43	// Create the client
44	LatticeClient := client.NewClient(
45		option.WithClientCredentials(clientId, clientSecret),
46		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
47		option.WithHTTPHeader(headers),
48	)
49
50	// Set the entity ID to listen for tasks
51	entityId := "<AGENT_ID>"
52	fmt.Printf("Streaming tasks for entity: %s...\n", entityId)
53
54	// Create context for the request
55	ctx := context.Background()
56
57	// Create agent stream request
58	agentStreamRequest := Lattice.AgentStreamRequest{
59		AgentSelector: &Lattice.EntityIDsSelector{
60			EntityIDs: []string{entityId},
61		},
62	}
63
64	// Stream tasks
65	stream, err := LatticeClient.Tasks.StreamAsAgent(ctx, &agentStreamRequest)
66	if err != nil {
67		fmt.Printf("Error streaming tasks: %v\n", err)
68		os.Exit(1)
69	}
70
71	// Process stream events
72	for {
73		select {
74		case <-ctx.Done():
75			log.Printf("Context canceled: %v", ctx.Err())
76			return
77		default:
78			// Continue processing
79		}
80
81		event, err := stream.Recv()
82
83		if errors.Is(err, io.EOF) {
84			log.Println("Stream completed successfully.")
85			return
86		}
87
88		if err != nil {
89			log.Printf("Error receiving message: %v", err)
90			continue
91		}
92
93		if event.Event == "heartbeat" {
94			timestamp := *event.Heartbeat.Timestamp
95			log.Printf("Heartbeat: %s", timestamp)
96		} else {
97			request := event.GetAgentRequest()
98			if executeRequest := request.GetExecuteRequest(); executeRequest != nil {
99				task := executeRequest.GetTask()
100				if task != nil {
101					taskId := *task.GetVersion().GetTaskID()
102					taskStatusVersion := *task.GetVersion().GetStatusVersion()
103					description := *task.GetDescription()
104
105					log.Printf("Starting task %s, version %d: %s", taskId, taskStatusVersion, description)
106
107					// Update task status to STATUS_EXECUTING
108					result, err := executeTask(ctx, LatticeClient, taskId, int(taskStatusVersion), entityId)
109					if err != nil {
110						log.Printf("Error starting task: %v", err)
111						continue
112					}
113
114					log.Printf("Started task with status version: %d", *result.StatusVersion)
115				}
116			} else if completeRequest := request.GetCompleteRequest(); completeRequest != nil {
117				taskToComplete := completeRequest.GetTaskID()
118				if taskToComplete != nil {
119					log.Printf("Completing task: %s", *taskToComplete)
120					err := completeTask(ctx, LatticeClient, *taskToComplete, entityId)
121					if err != nil {
122						log.Printf("Error completing task: %v", err)
123					}
124				}
125			} else if cancelRequest := request.GetCancelRequest(); cancelRequest != nil {
126				taskToCancel := cancelRequest.GetTaskID()
127				if taskToCancel != nil {
128					log.Printf("Cancelling task: %s", *taskToCancel)
129					err := cancelTask(ctx, LatticeClient, *taskToCancel, entityId)
130					if err != nil {
131						log.Printf("Error cancelling task: %v", err)
132					}
133				}
134			}
135		}
136
137		// Sleep briefly to prevent tight looping
138		time.Sleep(100 * time.Millisecond)
139	}
140}
141
142// executeTask updates the task status to STATUS_EXECUTING
143func executeTask(ctx context.Context, client *client.Client, taskId string, taskStatusVersion int, agentEntityId string) (*Lattice.TaskVersion, error) {
144	// Increment status version for the update
145	taskStatusVersion++
146
147	// Create system principal with the agent entity ID
148	principal := Lattice.Principal{
149		System: &Lattice.System{
150			EntityID: &agentEntityId,
151		},
152	}
153
154	taskStatus := Lattice.TaskStatus{
155		Status: Lattice.TaskStatusStatusStatusExecuting.Ptr(),
156	}
157
158	// Create task status update request
159	taskStatusUpdate := Lattice.TaskStatusUpdate{
160		TaskID:        taskId,
161		StatusVersion: &taskStatusVersion,
162		NewStatus:     &taskStatus,
163		Author:        &principal,
164	}
165
166	// Call the UpdateTaskStatus API
167	task, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
168	if err != nil {
169		return nil, fmt.Errorf("error updating task status: %w", err)
170	}
171
172	return task.Version, nil
173}
174
175// cancelTask handles cancellation requests from Lattice
176func cancelTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
177	// Get current task to retrieve status_version
178	getTaskRequest := Lattice.GetTaskRequest{
179		TaskID: taskId,
180	}
181	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
182	if err != nil {
183		return fmt.Errorf("error getting task: %w", err)
184	}
185	if task.Status == nil || task.Version == nil || task.Version.StatusVersion == nil {
186		return fmt.Errorf("task status or version is missing")
187	}
188	currentTaskStatus := task.Status.Status
189	taskStatusVersion := *task.Version.StatusVersion
190	taskStatusVersion++
191
192	// Create system principal with the agent entity ID
193	principal := Lattice.Principal{
194		System: &Lattice.System{
195			EntityID: &entityId,
196		},
197	}
198
199	if taskActive {
200		// Reject cancellation: task is active and cannot be cancelled
201		rejectedMessage := "Task is already active, and cannot be cancelled."
202		taskStatus := Lattice.TaskStatus{
203			// Because the cancellation is being rejected, we do not
204			// change the task status.
205			Status: currentTaskStatus,
206			TaskError: &Lattice.TaskError{
207				Code:    Lattice.TaskErrorCodeErrorCodeRejected.Ptr(),
208				Message: &rejectedMessage,
209			},
210		}
211
212		taskStatusUpdate := Lattice.TaskStatusUpdate{
213			TaskID:        taskId,
214			StatusVersion: &taskStatusVersion,
215			NewStatus:     &taskStatus,
216			Author:        &principal,
217		}
218
219		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
220		if err != nil {
221			return fmt.Errorf("error updating task status: %w", err)
222		}
223
224		log.Println("Task could not be cancelled.")
225	} else {
226		// Accept cancellation
227		cancelledMessage := "Task cancelled by agent."
228		taskStatus := Lattice.TaskStatus{
229			Status: Lattice.TaskStatusStatusStatusDoneNotOk.Ptr(),
230			TaskError: &Lattice.TaskError{
231				Code:    Lattice.TaskErrorCodeErrorCodeCancelled.Ptr(),
232				Message: &cancelledMessage,
233			},
234		}
235
236		taskStatusUpdate := Lattice.TaskStatusUpdate{
237			TaskID:        taskId,
238			StatusVersion: &taskStatusVersion,
239			NewStatus:     &taskStatus,
240			Author:        &principal,
241		}
242
243		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
244		if err != nil {
245			return fmt.Errorf("error updating task status: %w", err)
246		}
247
248		log.Println("Task has been cancelled.")
249	}
250
251	return nil
252}
253
254// completeTask updates the task status to STATUS_DONE_OK
255func completeTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
256	// Get current task to retrieve status_version
257	getTaskRequest := Lattice.GetTaskRequest{
258		TaskID: taskId,
259	}
260	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
261	if err != nil {
262		return fmt.Errorf("error getting task: %w", err)
263	}
264
265	if task.Version == nil || task.Version.StatusVersion == nil {
266		return fmt.Errorf("task version is missing")
267	}
268
269	taskStatusVersion := *task.Version.StatusVersion
270	// Increment version and update to terminal state
271	taskStatusVersion++
272
273	// Create system principal with the agent entity ID
274	principal := Lattice.Principal{
275		System: &Lattice.System{
276			EntityID: &entityId,
277		},
278	}
279
280	taskStatus := Lattice.TaskStatus{
281		Status: Lattice.TaskStatusStatusStatusDoneOk.Ptr(),
282	}
283
284	taskStatusUpdate := Lattice.TaskStatusUpdate{
285		TaskID:        taskId,
286		StatusVersion: &taskStatusVersion,
287		NewStatus:     &taskStatus,
288		Author:        &principal,
289	}
290
291	_, err = client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
292	if err != nil {
293		return fmt.Errorf("error updating task status: %w", err)
294	}
295
296	return nil
297}

The StreamAsAgent API establishes a server-sent events (SSE) stream that delivers three types of requests: executeRequest which notifies the agent to start executing a task, completeRequest which requests the agent to complete a task, and cancelRequest which requests the agent to cancel a task.

2

Handle cancellation requests

When a cancelRequest arrives, the agent retrieves the current task state and decides whether to accept or reject the cancellation.

Rejecting cancellation:

If the task is active and cannot be cancelled, the agent rejects the cancellation by keeping the current status and attaching a TaskError with code ERROR_CODE_REJECTED:

1package main
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"log"
9	"net/http"
10	"os"
11	"strings"
12	"time"
13
14	Lattice "github.com/anduril/lattice-sdk-go/v4"
15	"github.com/anduril/lattice-sdk-go/v4/client"
16	"github.com/anduril/lattice-sdk-go/v4/option"
17)
18
19// Sets whether the task is currently being processed. If so, it cannot be cancelled.
20var taskActive bool
21
22func main() {
23	// Get environment variables
24	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
25	clientSecret := os.Getenv("LATTICE_CLIENT_SECRET")
26	clientId := os.Getenv("LATTICE_CLIENT_ID")
27	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
28	taskActiveStr := os.Getenv("TASK_ACTIVE")
29
30	// Check required environment variables
31	if latticeEndpoint == "" || clientId == "" || clientSecret == "" || sandboxesToken == "" {
32		fmt.Println("Missing required environment variables")
33		os.Exit(1)
34	}
35
36	// Parse TASK_ACTIVE environment variable
37	taskActive = strings.ToLower(taskActiveStr) == "true"
38
39	// Initialize headers for sandbox authorization
40	headers := http.Header{}
41	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
42
43	// Create the client
44	LatticeClient := client.NewClient(
45		option.WithClientCredentials(clientId, clientSecret),
46		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
47		option.WithHTTPHeader(headers),
48	)
49
50	// Set the entity ID to listen for tasks
51	entityId := "<AGENT_ID>"
52	fmt.Printf("Streaming tasks for entity: %s...\n", entityId)
53
54	// Create context for the request
55	ctx := context.Background()
56
57	// Create agent stream request
58	agentStreamRequest := Lattice.AgentStreamRequest{
59		AgentSelector: &Lattice.EntityIDsSelector{
60			EntityIDs: []string{entityId},
61		},
62	}
63
64	// Stream tasks
65	stream, err := LatticeClient.Tasks.StreamAsAgent(ctx, &agentStreamRequest)
66	if err != nil {
67		fmt.Printf("Error streaming tasks: %v\n", err)
68		os.Exit(1)
69	}
70
71	// Process stream events
72	for {
73		select {
74		case <-ctx.Done():
75			log.Printf("Context canceled: %v", ctx.Err())
76			return
77		default:
78			// Continue processing
79		}
80
81		event, err := stream.Recv()
82
83		if errors.Is(err, io.EOF) {
84			log.Println("Stream completed successfully.")
85			return
86		}
87
88		if err != nil {
89			log.Printf("Error receiving message: %v", err)
90			continue
91		}
92
93		if event.Event == "heartbeat" {
94			timestamp := *event.Heartbeat.Timestamp
95			log.Printf("Heartbeat: %s", timestamp)
96		} else {
97			request := event.GetAgentRequest()
98			if executeRequest := request.GetExecuteRequest(); executeRequest != nil {
99				task := executeRequest.GetTask()
100				if task != nil {
101					taskId := *task.GetVersion().GetTaskID()
102					taskStatusVersion := *task.GetVersion().GetStatusVersion()
103					description := *task.GetDescription()
104
105					log.Printf("Starting task %s, version %d: %s", taskId, taskStatusVersion, description)
106
107					// Update task status to STATUS_EXECUTING
108					result, err := executeTask(ctx, LatticeClient, taskId, int(taskStatusVersion), entityId)
109					if err != nil {
110						log.Printf("Error starting task: %v", err)
111						continue
112					}
113
114					log.Printf("Started task with status version: %d", *result.StatusVersion)
115				}
116			} else if completeRequest := request.GetCompleteRequest(); completeRequest != nil {
117				taskToComplete := completeRequest.GetTaskID()
118				if taskToComplete != nil {
119					log.Printf("Completing task: %s", *taskToComplete)
120					err := completeTask(ctx, LatticeClient, *taskToComplete, entityId)
121					if err != nil {
122						log.Printf("Error completing task: %v", err)
123					}
124				}
125			} else if cancelRequest := request.GetCancelRequest(); cancelRequest != nil {
126				taskToCancel := cancelRequest.GetTaskID()
127				if taskToCancel != nil {
128					log.Printf("Cancelling task: %s", *taskToCancel)
129					err := cancelTask(ctx, LatticeClient, *taskToCancel, entityId)
130					if err != nil {
131						log.Printf("Error cancelling task: %v", err)
132					}
133				}
134			}
135		}
136
137		// Sleep briefly to prevent tight looping
138		time.Sleep(100 * time.Millisecond)
139	}
140}
141
142// executeTask updates the task status to STATUS_EXECUTING
143func executeTask(ctx context.Context, client *client.Client, taskId string, taskStatusVersion int, agentEntityId string) (*Lattice.TaskVersion, error) {
144	// Increment status version for the update
145	taskStatusVersion++
146
147	// Create system principal with the agent entity ID
148	principal := Lattice.Principal{
149		System: &Lattice.System{
150			EntityID: &agentEntityId,
151		},
152	}
153
154	taskStatus := Lattice.TaskStatus{
155		Status: Lattice.TaskStatusStatusStatusExecuting.Ptr(),
156	}
157
158	// Create task status update request
159	taskStatusUpdate := Lattice.TaskStatusUpdate{
160		TaskID:        taskId,
161		StatusVersion: &taskStatusVersion,
162		NewStatus:     &taskStatus,
163		Author:        &principal,
164	}
165
166	// Call the UpdateTaskStatus API
167	task, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
168	if err != nil {
169		return nil, fmt.Errorf("error updating task status: %w", err)
170	}
171
172	return task.Version, nil
173}
174
175// cancelTask handles cancellation requests from Lattice
176func cancelTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
177	// Get current task to retrieve status_version
178	getTaskRequest := Lattice.GetTaskRequest{
179		TaskID: taskId,
180	}
181	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
182	if err != nil {
183		return fmt.Errorf("error getting task: %w", err)
184	}
185	if task.Status == nil || task.Version == nil || task.Version.StatusVersion == nil {
186		return fmt.Errorf("task status or version is missing")
187	}
188	currentTaskStatus := task.Status.Status
189	taskStatusVersion := *task.Version.StatusVersion
190	taskStatusVersion++
191
192	// Create system principal with the agent entity ID
193	principal := Lattice.Principal{
194		System: &Lattice.System{
195			EntityID: &entityId,
196		},
197	}
198
199	if taskActive {
200		// Reject cancellation: task is active and cannot be cancelled
201		rejectedMessage := "Task is already active, and cannot be cancelled."
202		taskStatus := Lattice.TaskStatus{
203			// Because the cancellation is being rejected, we do not
204			// change the task status.
205			Status: currentTaskStatus,
206			TaskError: &Lattice.TaskError{
207				Code:    Lattice.TaskErrorCodeErrorCodeRejected.Ptr(),
208				Message: &rejectedMessage,
209			},
210		}
211
212		taskStatusUpdate := Lattice.TaskStatusUpdate{
213			TaskID:        taskId,
214			StatusVersion: &taskStatusVersion,
215			NewStatus:     &taskStatus,
216			Author:        &principal,
217		}
218
219		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
220		if err != nil {
221			return fmt.Errorf("error updating task status: %w", err)
222		}
223
224		log.Println("Task could not be cancelled.")
225	} else {
226		// Accept cancellation
227		cancelledMessage := "Task cancelled by agent."
228		taskStatus := Lattice.TaskStatus{
229			Status: Lattice.TaskStatusStatusStatusDoneNotOk.Ptr(),
230			TaskError: &Lattice.TaskError{
231				Code:    Lattice.TaskErrorCodeErrorCodeCancelled.Ptr(),
232				Message: &cancelledMessage,
233			},
234		}
235
236		taskStatusUpdate := Lattice.TaskStatusUpdate{
237			TaskID:        taskId,
238			StatusVersion: &taskStatusVersion,
239			NewStatus:     &taskStatus,
240			Author:        &principal,
241		}
242
243		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
244		if err != nil {
245			return fmt.Errorf("error updating task status: %w", err)
246		}
247
248		log.Println("Task has been cancelled.")
249	}
250
251	return nil
252}
253
254// completeTask updates the task status to STATUS_DONE_OK
255func completeTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
256	// Get current task to retrieve status_version
257	getTaskRequest := Lattice.GetTaskRequest{
258		TaskID: taskId,
259	}
260	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
261	if err != nil {
262		return fmt.Errorf("error getting task: %w", err)
263	}
264
265	if task.Version == nil || task.Version.StatusVersion == nil {
266		return fmt.Errorf("task version is missing")
267	}
268
269	taskStatusVersion := *task.Version.StatusVersion
270	// Increment version and update to terminal state
271	taskStatusVersion++
272
273	// Create system principal with the agent entity ID
274	principal := Lattice.Principal{
275		System: &Lattice.System{
276			EntityID: &entityId,
277		},
278	}
279
280	taskStatus := Lattice.TaskStatus{
281		Status: Lattice.TaskStatusStatusStatusDoneOk.Ptr(),
282	}
283
284	taskStatusUpdate := Lattice.TaskStatusUpdate{
285		TaskID:        taskId,
286		StatusVersion: &taskStatusVersion,
287		NewStatus:     &taskStatus,
288		Author:        &principal,
289	}
290
291	_, err = client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
292	if err != nil {
293		return fmt.Errorf("error updating task status: %w", err)
294	}
295
296	return nil
297}

Accepting cancellation:

If the task can be cancelled, the agent accepts by setting the status to STATUS_DONE_NOT_OK with a TaskError indicating ERROR_CODE_CANCELLED:

1package main
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"io"
8	"log"
9	"net/http"
10	"os"
11	"strings"
12	"time"
13
14	Lattice "github.com/anduril/lattice-sdk-go/v4"
15	"github.com/anduril/lattice-sdk-go/v4/client"
16	"github.com/anduril/lattice-sdk-go/v4/option"
17)
18
19// Sets whether the task is currently being processed. If so, it cannot be cancelled.
20var taskActive bool
21
22func main() {
23	// Get environment variables
24	latticeEndpoint := os.Getenv("LATTICE_ENDPOINT")
25	clientSecret := os.Getenv("LATTICE_CLIENT_SECRET")
26	clientId := os.Getenv("LATTICE_CLIENT_ID")
27	sandboxesToken := os.Getenv("SANDBOXES_TOKEN")
28	taskActiveStr := os.Getenv("TASK_ACTIVE")
29
30	// Check required environment variables
31	if latticeEndpoint == "" || clientId == "" || clientSecret == "" || sandboxesToken == "" {
32		fmt.Println("Missing required environment variables")
33		os.Exit(1)
34	}
35
36	// Parse TASK_ACTIVE environment variable
37	taskActive = strings.ToLower(taskActiveStr) == "true"
38
39	// Initialize headers for sandbox authorization
40	headers := http.Header{}
41	headers.Add("Anduril-Sandbox-Authorization", fmt.Sprintf("Bearer %s", sandboxesToken))
42
43	// Create the client
44	LatticeClient := client.NewClient(
45		option.WithClientCredentials(clientId, clientSecret),
46		option.WithBaseURL(fmt.Sprintf("https://%s", latticeEndpoint)),
47		option.WithHTTPHeader(headers),
48	)
49
50	// Set the entity ID to listen for tasks
51	entityId := "<AGENT_ID>"
52	fmt.Printf("Streaming tasks for entity: %s...\n", entityId)
53
54	// Create context for the request
55	ctx := context.Background()
56
57	// Create agent stream request
58	agentStreamRequest := Lattice.AgentStreamRequest{
59		AgentSelector: &Lattice.EntityIDsSelector{
60			EntityIDs: []string{entityId},
61		},
62	}
63
64	// Stream tasks
65	stream, err := LatticeClient.Tasks.StreamAsAgent(ctx, &agentStreamRequest)
66	if err != nil {
67		fmt.Printf("Error streaming tasks: %v\n", err)
68		os.Exit(1)
69	}
70
71	// Process stream events
72	for {
73		select {
74		case <-ctx.Done():
75			log.Printf("Context canceled: %v", ctx.Err())
76			return
77		default:
78			// Continue processing
79		}
80
81		event, err := stream.Recv()
82
83		if errors.Is(err, io.EOF) {
84			log.Println("Stream completed successfully.")
85			return
86		}
87
88		if err != nil {
89			log.Printf("Error receiving message: %v", err)
90			continue
91		}
92
93		if event.Event == "heartbeat" {
94			timestamp := *event.Heartbeat.Timestamp
95			log.Printf("Heartbeat: %s", timestamp)
96		} else {
97			request := event.GetAgentRequest()
98			if executeRequest := request.GetExecuteRequest(); executeRequest != nil {
99				task := executeRequest.GetTask()
100				if task != nil {
101					taskId := *task.GetVersion().GetTaskID()
102					taskStatusVersion := *task.GetVersion().GetStatusVersion()
103					description := *task.GetDescription()
104
105					log.Printf("Starting task %s, version %d: %s", taskId, taskStatusVersion, description)
106
107					// Update task status to STATUS_EXECUTING
108					result, err := executeTask(ctx, LatticeClient, taskId, int(taskStatusVersion), entityId)
109					if err != nil {
110						log.Printf("Error starting task: %v", err)
111						continue
112					}
113
114					log.Printf("Started task with status version: %d", *result.StatusVersion)
115				}
116			} else if completeRequest := request.GetCompleteRequest(); completeRequest != nil {
117				taskToComplete := completeRequest.GetTaskID()
118				if taskToComplete != nil {
119					log.Printf("Completing task: %s", *taskToComplete)
120					err := completeTask(ctx, LatticeClient, *taskToComplete, entityId)
121					if err != nil {
122						log.Printf("Error completing task: %v", err)
123					}
124				}
125			} else if cancelRequest := request.GetCancelRequest(); cancelRequest != nil {
126				taskToCancel := cancelRequest.GetTaskID()
127				if taskToCancel != nil {
128					log.Printf("Cancelling task: %s", *taskToCancel)
129					err := cancelTask(ctx, LatticeClient, *taskToCancel, entityId)
130					if err != nil {
131						log.Printf("Error cancelling task: %v", err)
132					}
133				}
134			}
135		}
136
137		// Sleep briefly to prevent tight looping
138		time.Sleep(100 * time.Millisecond)
139	}
140}
141
142// executeTask updates the task status to STATUS_EXECUTING
143func executeTask(ctx context.Context, client *client.Client, taskId string, taskStatusVersion int, agentEntityId string) (*Lattice.TaskVersion, error) {
144	// Increment status version for the update
145	taskStatusVersion++
146
147	// Create system principal with the agent entity ID
148	principal := Lattice.Principal{
149		System: &Lattice.System{
150			EntityID: &agentEntityId,
151		},
152	}
153
154	taskStatus := Lattice.TaskStatus{
155		Status: Lattice.TaskStatusStatusStatusExecuting.Ptr(),
156	}
157
158	// Create task status update request
159	taskStatusUpdate := Lattice.TaskStatusUpdate{
160		TaskID:        taskId,
161		StatusVersion: &taskStatusVersion,
162		NewStatus:     &taskStatus,
163		Author:        &principal,
164	}
165
166	// Call the UpdateTaskStatus API
167	task, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
168	if err != nil {
169		return nil, fmt.Errorf("error updating task status: %w", err)
170	}
171
172	return task.Version, nil
173}
174
175// cancelTask handles cancellation requests from Lattice
176func cancelTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
177	// Get current task to retrieve status_version
178	getTaskRequest := Lattice.GetTaskRequest{
179		TaskID: taskId,
180	}
181	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
182	if err != nil {
183		return fmt.Errorf("error getting task: %w", err)
184	}
185	if task.Status == nil || task.Version == nil || task.Version.StatusVersion == nil {
186		return fmt.Errorf("task status or version is missing")
187	}
188	currentTaskStatus := task.Status.Status
189	taskStatusVersion := *task.Version.StatusVersion
190	taskStatusVersion++
191
192	// Create system principal with the agent entity ID
193	principal := Lattice.Principal{
194		System: &Lattice.System{
195			EntityID: &entityId,
196		},
197	}
198
199	if taskActive {
200		// Reject cancellation: task is active and cannot be cancelled
201		rejectedMessage := "Task is already active, and cannot be cancelled."
202		taskStatus := Lattice.TaskStatus{
203			// Because the cancellation is being rejected, we do not
204			// change the task status.
205			Status: currentTaskStatus,
206			TaskError: &Lattice.TaskError{
207				Code:    Lattice.TaskErrorCodeErrorCodeRejected.Ptr(),
208				Message: &rejectedMessage,
209			},
210		}
211
212		taskStatusUpdate := Lattice.TaskStatusUpdate{
213			TaskID:        taskId,
214			StatusVersion: &taskStatusVersion,
215			NewStatus:     &taskStatus,
216			Author:        &principal,
217		}
218
219		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
220		if err != nil {
221			return fmt.Errorf("error updating task status: %w", err)
222		}
223
224		log.Println("Task could not be cancelled.")
225	} else {
226		// Accept cancellation
227		cancelledMessage := "Task cancelled by agent."
228		taskStatus := Lattice.TaskStatus{
229			Status: Lattice.TaskStatusStatusStatusDoneNotOk.Ptr(),
230			TaskError: &Lattice.TaskError{
231				Code:    Lattice.TaskErrorCodeErrorCodeCancelled.Ptr(),
232				Message: &cancelledMessage,
233			},
234		}
235
236		taskStatusUpdate := Lattice.TaskStatusUpdate{
237			TaskID:        taskId,
238			StatusVersion: &taskStatusVersion,
239			NewStatus:     &taskStatus,
240			Author:        &principal,
241		}
242
243		_, err := client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
244		if err != nil {
245			return fmt.Errorf("error updating task status: %w", err)
246		}
247
248		log.Println("Task has been cancelled.")
249	}
250
251	return nil
252}
253
254// completeTask updates the task status to STATUS_DONE_OK
255func completeTask(ctx context.Context, client *client.Client, taskId string, entityId string) error {
256	// Get current task to retrieve status_version
257	getTaskRequest := Lattice.GetTaskRequest{
258		TaskID: taskId,
259	}
260	task, err := client.Tasks.GetTask(ctx, &getTaskRequest)
261	if err != nil {
262		return fmt.Errorf("error getting task: %w", err)
263	}
264
265	if task.Version == nil || task.Version.StatusVersion == nil {
266		return fmt.Errorf("task version is missing")
267	}
268
269	taskStatusVersion := *task.Version.StatusVersion
270	// Increment version and update to terminal state
271	taskStatusVersion++
272
273	// Create system principal with the agent entity ID
274	principal := Lattice.Principal{
275		System: &Lattice.System{
276			EntityID: &entityId,
277		},
278	}
279
280	taskStatus := Lattice.TaskStatus{
281		Status: Lattice.TaskStatusStatusStatusDoneOk.Ptr(),
282	}
283
284	taskStatusUpdate := Lattice.TaskStatusUpdate{
285		TaskID:        taskId,
286		StatusVersion: &taskStatusVersion,
287		NewStatus:     &taskStatus,
288		Author:        &principal,
289	}
290
291	_, err = client.Tasks.UpdateTaskStatus(ctx, &taskStatusUpdate)
292	if err != nil {
293		return fmt.Errorf("error updating task status: %w", err)
294	}
295
296	return nil
297}

The agent must first retrieve the current task using GetTask to obtain the current statusVersion, then increment it before calling UpdateTaskStatus.

What’s next